Data track schema metadata

src/api/SignalClient.ts

@@ -56,7 +56,11 @@ import {
 } from '@livekit/protocol';
 import log, { LoggerNames, getLogger } from '../logger';
 import type { DataTrackHandle } from '../room/data-track/handle';
-import { type DataTrackSid } from '../room/data-track/types';
+import {
+  DataTrackFrameEncoding,
+  DataTrackSchemaId,
+  type DataTrackSid,
+} from '../room/data-track/types';
 import { ConnectionError } from '../room/errors';
 import CriticalTimers from '../room/timers';
 import type { LoggerOptions } from '../room/types';
@@ -744,13 +748,21 @@ export class SignalClient {
     });
   }
 
-  sendPublishDataTrackRequest(handle: DataTrackHandle, name: string, usesE2ee: boolean) {
+  sendPublishDataTrackRequest(
+    handle: DataTrackHandle,
+    name: string,
+    usesE2ee: boolean,
+    schema?: DataTrackSchemaId,
+    frameEncoding?: DataTrackFrameEncoding,
+  ) {
     return this.sendRequest({
       case: 'publishDataTrackRequest',
       value: new PublishDataTrackRequest({
         pubHandle: handle,
         name: name,
         encryption: usesE2ee ? Encryption_Type.GCM : Encryption_Type.NONE,
+        schema: schema ? DataTrackSchemaId.toProtobuf(schema) : undefined,
+        frameEncoding: frameEncoding ? DataTrackFrameEncoding.toProtobuf(frameEncoding) : undefined,
       }),
     });
   }

src/index.ts

@@ -180,6 +180,11 @@ export type {
   DataTrackSubscribeOptions,
   RemoteDataTrackPipelineOptions,
 };
+export type {
+  DataTrackSchemaId,
+  DataTrackSchemaEncoding,
+  DataTrackFrameEncoding,
+} from './room/data-track/types';
 export { DataTrackPacket, type DataTrackPacketHeader } from './room/data-track/packet';
 export {
   type DataTrackExtensions,

src/room/Room.ts

@@ -305,7 +305,13 @@ class Room extends (EventEmitter as new () => TypedEmitter<RoomEventCallbacks>)
     this.outgoingDataTrackManager = new OutgoingDataTrackManager({ e2eeManager: this.e2eeManager });
     this.outgoingDataTrackManager
       .on('sfuPublishRequest', (event) => {
-        this.engine.client.sendPublishDataTrackRequest(event.handle, event.name, event.usesE2ee);
+        this.engine.client.sendPublishDataTrackRequest(
+          event.handle,
+          event.name,
+          event.usesE2ee,
+          event.schema,
+          event.frameEncoding,
+        );
       })
       .on('sfuUnpublishRequest', (event) => {
         this.engine.client.sendUnPublishDataTrackRequest(event.handle);
@@ -687,12 +693,7 @@ class Room extends (EventEmitter as new () => TypedEmitter<RoomEventCallbacks>)
 
         this.outgoingDataTrackManager.receivedSfuPublishResponse(event.info.pubHandle, {
           type: 'ok',
-          data: {
-            sid: event.info.sid,
-            pubHandle: event.info.pubHandle,
-            name: event.info.name,
-            usesE2ee: event.info.encryption !== Encryption_Type.NONE,
-          },
+          data: DataTrackInfo.from(event.info),
         });
       })
       .on(EngineEvent.UnPublishDataTrackResponse, (event) => {

src/room/data-track/outgoing/OutgoingDataTrackManager.test.ts

@@ -11,6 +11,7 @@ import RTCEngine from '../../RTCEngine';
 import Room from '../../Room';
 import { DataTrackHandle } from '../handle';
 import { DataTrackPacket, FrameMarker } from '../packet';
+import { DataTrackSchemaError } from '../schema';
 import OutgoingDataTrackManager, {
   type DataTrackOutgoingManagerCallbacks,
   Descriptor,
@@ -132,6 +133,27 @@ describe('DataTrackOutgoingManager', () => {
     );
   });
 
+  it('should reject publishing when schema metadata is invalid', async () => {
+    const manager = new OutgoingDataTrackManager();
+    const managerEvents = subscribeToEvents<DataTrackOutgoingManagerCallbacks>(manager, [
+      'sfuPublishRequest',
+    ]);
+
+    // Providing a schema ID without a frame encoding is invalid.
+    const localDataTrack = new LocalDataTrack(
+      { name: 'test', schema: { name: 'my_schema', encoding: 'jsonSchema' } },
+      manager,
+    );
+
+    await expect(localDataTrack.publish()).rejects.toStrictEqual(
+      DataTrackPublishError.invalidSchema(DataTrackSchemaError.missingFrameEncoding()),
+    );
+
+    // The invalid request must not be sent to the SFU.
+    expect(managerEvents.areThereBufferedEvents('sfuPublishRequest')).toBe(false);
+    expect(localDataTrack.isPublished()).toStrictEqual(false);
+  });
+
   it('should test track publishing (cancellation half way through)', async () => {
     const manager = new OutgoingDataTrackManager();
     const managerEvents = subscribeToEvents<DataTrackOutgoingManagerCallbacks>(manager, [
@@ -251,6 +273,65 @@ describe('DataTrackOutgoingManager', () => {
     expect(handle).not.toStrictEqual(handle2);
   });
 
+  it.each([
+    {
+      title: 'well-known encodings',
+      schema: { name: 'my_schema', encoding: 'jsonSchema' },
+      frameEncoding: 'json',
+    },
+    {
+      title: 'custom encodings',
+      schema: { name: 'my_schema', encoding: { custom: 'a' } },
+      frameEncoding: { custom: 'b' },
+    },
+  ] as const)(
+    'should forward schema and frame encoding on publish ($title)',
+    async ({ schema, frameEncoding }) => {
+      const manager = new OutgoingDataTrackManager();
+      const managerEvents = subscribeToEvents<DataTrackOutgoingManagerCallbacks>(manager, [
+        'sfuPublishRequest',
+      ]);
+
+      const localDataTrack = new LocalDataTrack({ name: 'test', schema, frameEncoding }, manager);
+
+      // 1. Publish a data track with schema metadata
+      const publishRequestPromise = localDataTrack.publish();
+
+      // 2. The publish request sent to the SFU carries the schema and frame encoding
+      const sfuPublishEvent = await managerEvents.waitFor('sfuPublishRequest');
+      expect(sfuPublishEvent.schema).toStrictEqual(schema);
+      expect(sfuPublishEvent.frameEncoding).toStrictEqual(frameEncoding);
+      const handle = sfuPublishEvent.handle;
+
+      // 3. Respond as the SFU would, echoing the metadata back on the DataTrackInfo
+      manager.receivedSfuPublishResponse(handle, {
+        type: 'ok',
+        data: {
+          sid: 'bogus-sid',
+          pubHandle: handle,
+          name: 'test',
+          usesE2ee: false,
+          schema,
+          frameEncoding,
+        },
+      });
+      await publishRequestPromise;
+
+      // 4. The metadata is reflected on the local track's info
+      expect(localDataTrack.info?.schema).toStrictEqual(schema);
+      expect(localDataTrack.info?.frameEncoding).toStrictEqual(frameEncoding);
+
+      // 5. Republishing after a full reconnect re-announces the track with the same metadata,
+      // since a track's schema is immutable even though it is assigned a new sid.
+      manager.sfuWillRepublishTracks();
+
+      const republishEvent = await managerEvents.waitFor('sfuPublishRequest');
+      expect(republishEvent.handle).toStrictEqual(handle);
+      expect(republishEvent.schema).toStrictEqual(schema);
+      expect(republishEvent.frameEncoding).toStrictEqual(frameEncoding);
+    },
+  );
+
   it.each([
     // Single packet payload case
     [

src/room/data-track/outgoing/OutgoingDataTrackManager.ts

@@ -8,6 +8,7 @@ import { Future } from '../../utils';
 import LocalDataTrack from '../LocalDataTrack';
 import type { DataTrackFrameInternal } from '../frame';
 import { DataTrackHandle, DataTrackHandleAllocator } from '../handle';
+import { type DataTrackSchemaError, validateSchemaMetadata } from '../schema';
 import { type DataTrackInfo } from '../types';
 import {
   DataTrackPublishError,
@@ -215,6 +216,12 @@ export default class OutgoingDataTrackManager extends (EventEmitter as new () =>
     options: DataTrackOptions,
     signal?: AbortSignal,
   ): Promise<Throws<DataTrackHandle, DataTrackPublishError>> {
+    try {
+      validateSchemaMetadata(options.frameEncoding, options.schema?.encoding);
+    } catch (error) {
+      throw DataTrackPublishError.invalidSchema(error as DataTrackSchemaError);
+    }
+
     const handle = this.handleAllocator.get();
     if (!handle) {
       throw DataTrackPublishError.limitReached();
@@ -263,6 +270,8 @@ export default class OutgoingDataTrackManager extends (EventEmitter as new () =>
       handle,
       name: options.name,
       usesE2ee: this.e2eeManager !== null,
+      schema: options.schema,
+      frameEncoding: options.frameEncoding,
     });
 
     await descriptor.completionFuture.promise;
@@ -397,6 +406,8 @@ export default class OutgoingDataTrackManager extends (EventEmitter as new () =>
             handle: descriptor.info.pubHandle,
             name: descriptor.info.name,
             usesE2ee: descriptor.info.usesE2ee,
+            schema: descriptor.info.schema,
+            frameEncoding: descriptor.info.frameEncoding,
           });
       }
     }

src/room/data-track/outgoing/errors.ts

@@ -1,5 +1,6 @@
 import { LivekitReasonedError } from '../../errors';
 import { DataTrackPacketizerError } from '../packetizer';
+import type { DataTrackSchemaError } from '../schema';
 
 export enum DataTrackPublishErrorReason {
   /**
@@ -30,6 +31,9 @@ export enum DataTrackPublishErrorReason {
   /** There was an error publishing, but it was not something that could be sorted into a known
    * category. */
   Unknown = 7,
+
+  /** Schema metadata is invalid. */
+  InvalidSchema = 8,
 }
 
 export class DataTrackPublishError<
@@ -108,6 +112,12 @@ export class DataTrackPublishError<
       DataTrackPublishErrorReason.Cancelled,
     );
   }
+
+  static invalidSchema(cause: DataTrackSchemaError) {
+    return new DataTrackPublishError(cause.message, DataTrackPublishErrorReason.InvalidSchema, {
+      cause,
+    });
+  }
 }
 
 export enum DataTrackPushFrameErrorReason {

src/room/data-track/outgoing/types.ts

@@ -1,12 +1,23 @@
 import type { NonSharedUint8Array } from '../../../type-polyfills/non-shared-typed-arrays';
 import type LocalDataTrack from '../LocalDataTrack';
 import { type DataTrackHandle } from '../handle';
-import { type DataTrackInfo, type DataTrackSid } from '../types';
+import {
+  type DataTrackFrameEncoding,
+  type DataTrackInfo,
+  type DataTrackSchemaId,
+  type DataTrackSid,
+} from '../types';
 import { type DataTrackPublishError, type DataTrackPublishErrorReason } from './errors';
 
 /** Options for publishing a data track. */
 export type DataTrackOptions = {
   name: string;
+
+  /** Schema describing frames sent on the track. */
+  schema?: DataTrackSchemaId;
+
+  /** Encoding of frames sent on the track. */
+  frameEncoding?: DataTrackFrameEncoding;
 };
 
 /** Encodes whether a data track publish request to the SFU has been successful or not. */
@@ -27,6 +38,8 @@ export type EventSfuPublishRequest = {
   handle: DataTrackHandle;
   name: string;
   usesE2ee: boolean;
+  schema?: DataTrackSchemaId;
+  frameEncoding?: DataTrackFrameEncoding;
 };
 
 /** Request sent to the SFU to unpublish a track. */