From 9f56242f722343391e1f2a220b80cba8829800a8 Mon Sep 17 00:00:00 2001 From: Jacob Gelman <3182119+ladvoc@users.noreply.github.com> Date: Thu, 3 Sep 2026 11:56:39 -0700 Subject: [PATCH 01/23] Add data track publish request conversions --- datatrack/errors.go | 28 +++++++++++++++++ datatrack/proto.go | 70 +++++++++++++++++++++++++++++++++++++++++ datatrack/proto_test.go | 26 +++++++++++++++ 3 files changed, 124 insertions(+) create mode 100644 datatrack/errors.go diff --git a/datatrack/errors.go b/datatrack/errors.go new file mode 100644 index 00000000..cf465473 --- /dev/null +++ b/datatrack/errors.go @@ -0,0 +1,28 @@ +// Copyright 2026 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 datatrack + +import "errors" + +var ( + ErrNotAllowed = errors.New("data track publishing unauthorized") + ErrDuplicateName = errors.New("track name already taken") + ErrInvalidName = errors.New("track name invalid") + ErrLimitReached = errors.New("data track publication limit reached") + ErrPublishTimeout = errors.New("timed out publishing data track") + ErrDisconnected = errors.New("room disconnected") + ErrUnpublished = errors.New("track unpublished") + ErrQueueFull = errors.New("queue full") +) diff --git a/datatrack/proto.go b/datatrack/proto.go index 3b438311..112f19f5 100644 --- a/datatrack/proto.go +++ b/datatrack/proto.go @@ -203,3 +203,73 @@ func extractTrackInfo(participant *livekit.ParticipantInfo) ([]Info, error) { } return infos, nil } + +// publishRequest is what the local participant asks the SFU to publish. +type publishRequest struct { + handle trackHandle + name string + usesE2EE bool + schema *SchemaID + frameEncoding FrameEncoding +} + +func (r publishRequest) toProto() *livekit.PublishDataTrackRequest { + req := &livekit.PublishDataTrackRequest{ + PubHandle: uint32(r.handle), + Name: r.name, + Encryption: livekit.Encryption_NONE, + } + if r.usesE2EE { + req.Encryption = livekit.Encryption_GCM + } + if r.schema != nil { + req.Schema = schemaIDToProto(*r.schema) + } + if r.frameEncoding != nil { + req.FrameEncoding = frameEncodingToProto(r.frameEncoding) + } + return req +} + +func unpublishRequestToProto(handle trackHandle) *livekit.UnpublishDataTrackRequest { + return &livekit.UnpublishDataTrackRequest{PubHandle: uint32(handle)} +} + +// publishResponse is the SFU's answer to a publishRequest: info when accepted, err when rejected. +type publishResponse struct { + handle trackHandle + info Info + err error +} + +// publishRejectionFromRequestResponse extracts a rejected publish request; ok is false when the +// response is not a publish rejection. +func publishRejectionFromRequestResponse(msg *livekit.RequestResponse) (rejection publishResponse, ok bool) { + request, isPublish := msg.GetRequest().(*livekit.RequestResponse_PublishDataTrack) + if !isPublish || msg.GetReason() == livekit.RequestResponse_OK { + return publishResponse{}, false + } + handle, err := handleFromUint32(request.PublishDataTrack.GetPubHandle()) + if err != nil { + return publishResponse{}, false + } + + rejection = publishResponse{handle: handle} + switch msg.GetReason() { + case livekit.RequestResponse_NOT_ALLOWED: + rejection.err = ErrNotAllowed + case livekit.RequestResponse_DUPLICATE_NAME: + rejection.err = ErrDuplicateName + case livekit.RequestResponse_INVALID_NAME: + rejection.err = ErrInvalidName + case livekit.RequestResponse_LIMIT_EXCEEDED: + rejection.err = ErrLimitReached + default: + rejection.err = fmt.Errorf("data track publish rejected (%s): %s", msg.GetReason(), msg.GetMessage()) + return rejection, true + } + if message := msg.GetMessage(); message != "" { + rejection.err = fmt.Errorf("%w: %s", rejection.err, message) + } + return rejection, true +} diff --git a/datatrack/proto_test.go b/datatrack/proto_test.go index f1e40cf6..116ab0c3 100644 --- a/datatrack/proto_test.go +++ b/datatrack/proto_test.go @@ -115,3 +115,29 @@ func TestProto_ExtractTrackInfo(t *testing.T) { require.Equal(t, "track1", infos[0].Name) require.Equal(t, livekit.TrackID("DTR_1234"), infos[0].SID) } + +func TestProto_PublishRequestToProto(t *testing.T) { + request := publishRequest{handle: 1, name: "track", usesE2EE: true}.toProto() + require.Equal(t, uint32(1), request.GetPubHandle()) + require.Equal(t, "track", request.GetName()) + require.Equal(t, livekit.Encryption_GCM, request.GetEncryption()) +} + +func TestProto_UnpublishRequestToProto(t *testing.T) { + request := unpublishRequestToProto(1) + require.Equal(t, uint32(1), request.GetPubHandle()) +} + +func TestProto_PublishRejectionFromRequestResponse(t *testing.T) { + response := &livekit.RequestResponse{ + Request: &livekit.RequestResponse_PublishDataTrack{ + PublishDataTrack: &livekit.PublishDataTrackRequest{PubHandle: 1}, + }, + Reason: livekit.RequestResponse_NOT_ALLOWED, + } + + rejection, ok := publishRejectionFromRequestResponse(response) + require.True(t, ok) + require.Equal(t, trackHandle(1), rejection.handle) + require.ErrorIs(t, rejection.err, ErrNotAllowed) +} From e859b33806bba870e340e50c36f825f11c5cb192 Mon Sep 17 00:00:00 2001 From: Jacob Gelman <3182119+ladvoc@users.noreply.github.com> Date: Thu, 3 Sep 2026 11:56:39 -0700 Subject: [PATCH 02/23] Add local data track manager --- datatrack/local.go | 450 ++++++++++++++++++++++++++++++++++++++++ datatrack/local_test.go | 321 ++++++++++++++++++++++++++++ 2 files changed, 771 insertions(+) create mode 100644 datatrack/local.go create mode 100644 datatrack/local_test.go diff --git a/datatrack/local.go b/datatrack/local.go new file mode 100644 index 00000000..346d2cd8 --- /dev/null +++ b/datatrack/local.go @@ -0,0 +1,450 @@ +// Copyright 2026 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 datatrack + +import ( + "context" + "sync" + "sync/atomic" + + dtp "github.com/livekit/protocol/datatrack" + "github.com/livekit/protocol/livekit" + "github.com/livekit/protocol/logger" +) + +// transportMTU is the largest packet sent over the data channel. +const transportMTU = 16_000 + +// LocalTransport carries what the local manager produces: signal requests to the SFU and frame +// packets over the data channel. +type LocalTransport interface { + SendPublishRequest(req *livekit.PublishDataTrackRequest) error + SendUnpublishRequest(req *livekit.UnpublishDataTrackRequest) error + // SendFrame queues all packets of one frame; it must not block. + SendFrame(packets [][]byte) +} + +// Encryptor seals frame payloads for end-to-end encryption. +type Encryptor interface { + Encrypt(payload []byte) ([]byte, E2EEExtension, error) +} + +// PublishOptions describe a data track to publish. +type PublishOptions struct { + Name string + Schema *SchemaID + FrameEncoding FrameEncoding +} + +// PublishOption customizes a publication. +type PublishOption func(*PublishOptions) + +// WithSchema associates a schema with the track's frames. +func WithSchema(schema SchemaID) PublishOption { + return func(options *PublishOptions) { + options.Schema = &schema + } +} + +// WithFrameEncoding declares the encoding of the track's frames. +func WithFrameEncoding(encoding FrameEncoding) PublishOption { + return func(options *PublishOptions) { + options.FrameEncoding = encoding + } +} + +type LocalManagerParams struct { + Transport LocalTransport + // Encryptor enables end-to-end encryption for every published track; nil disables it. + Encryptor Encryptor + Logger logger.Logger +} + +// LocalManager tracks the data tracks published by the local participant. Methods are safe for +// concurrent use. +type LocalManager struct { + params LocalManagerParams + + mu sync.Mutex + handles handleAllocator + pending map[trackHandle]chan publishResult + active map[trackHandle]*LocalTrack +} + +type publishResult struct { + track *LocalTrack + err error +} + +func NewLocalManager(params LocalManagerParams) *LocalManager { + if params.Logger == nil { + params.Logger = logger.GetLogger() + } + return &LocalManager{ + params: params, + pending: make(map[trackHandle]chan publishResult), + active: make(map[trackHandle]*LocalTrack), + } +} + +// Publish requests a publication and waits for the SFU's answer. Ending ctx abandons the request; +// should the SFU accept it afterwards, the track is unpublished right away. +func (m *LocalManager) Publish(ctx context.Context, options PublishOptions) (*LocalTrack, error) { + var schemaEncoding SchemaEncoding + if options.Schema != nil { + schemaEncoding = options.Schema.Encoding + if schemaEncoding == nil { + schemaEncoding = SchemaEncodingOther + } + } + if err := validateSchema(options.FrameEncoding, schemaEncoding); err != nil { + return nil, err + } + + m.mu.Lock() + handle, ok := m.handles.get() + if !ok { + m.mu.Unlock() + return nil, ErrLimitReached + } + result := make(chan publishResult, 1) + m.pending[handle] = result + m.mu.Unlock() + + request := publishRequest{ + handle: handle, + name: options.Name, + usesE2EE: m.params.Encryptor != nil, + schema: options.Schema, + frameEncoding: options.FrameEncoding, + } + if err := m.params.Transport.SendPublishRequest(request.toProto()); err != nil { + m.mu.Lock() + delete(m.pending, handle) + m.mu.Unlock() + return nil, err + } + + select { + case res := <-result: + return res.track, res.err + case <-ctx.Done(): + m.mu.Lock() + _, stillPending := m.pending[handle] + delete(m.pending, handle) + m.mu.Unlock() + if !stillPending { + // the answer came first and was delivered under the lock + if res := <-result; res.track != nil { + res.track.Unpublish() + } + } + return nil, ctx.Err() + } +} + +// HandlePublishResponse completes the publication the SFU accepted. +func (m *LocalManager) HandlePublishResponse(msg *livekit.PublishDataTrackResponse) { + info, err := infoFromProto(msg.GetInfo()) + if err != nil { + m.params.Logger.Warnw("ignoring invalid publish data track response", err) + return + } + m.resolvePublish(publishResponse{handle: info.pubHandle, info: info}) +} + +// HandleRequestResponse fails the publication a rejection refers to. It reports whether the +// response was a publish rejection. +func (m *LocalManager) HandleRequestResponse(msg *livekit.RequestResponse) bool { + rejection, ok := publishRejectionFromRequestResponse(msg) + if !ok { + return false + } + m.resolvePublish(rejection) + return true +} + +func (m *LocalManager) resolvePublish(response publishResponse) { + m.mu.Lock() + if result, ok := m.pending[response.handle]; ok { + delete(m.pending, response.handle) + res := publishResult{err: response.err} + if response.err == nil { + res.track = newLocalTrack(m, response.info) + m.active[response.handle] = res.track + } + result <- res + m.mu.Unlock() + return + } + track, isActive := m.active[response.handle] + m.mu.Unlock() + + switch { + case isActive && response.err != nil: + m.params.Logger.Warnw("republish failed for data track", response.err, "handle", response.handle) + case isActive: + if track.completeRepublish(response.info.SID) { + m.params.Logger.Debugw("data track republished", "handle", response.handle, "sid", response.info.SID) + } else { + m.params.Logger.Warnw("data track already published", nil, "handle", response.handle) + } + case response.err == nil: + // accepted after the request was abandoned, unpublish to keep the SFU consistent + m.sendUnpublish(response.handle) + } +} + +// HandleUnpublishResponse completes an unpublication, whether requested locally or by the SFU. +func (m *LocalManager) HandleUnpublishResponse(msg *livekit.UnpublishDataTrackResponse) { + handle, err := handleFromUint32(msg.GetInfo().GetPubHandle()) + if err != nil { + m.params.Logger.Warnw("ignoring invalid unpublish data track response", err) + return + } + + m.mu.Lock() + if result, ok := m.pending[handle]; ok { + delete(m.pending, handle) + result <- publishResult{err: ErrUnpublished} + } + track, ok := m.active[handle] + delete(m.active, handle) + m.mu.Unlock() + + if ok { + track.markUnpublished() + } +} + +func (m *LocalManager) unpublish(handle trackHandle) { + m.mu.Lock() + track, ok := m.active[handle] + delete(m.active, handle) + m.mu.Unlock() + if !ok { + return + } + track.markUnpublished() + m.sendUnpublish(handle) +} + +func (m *LocalManager) sendUnpublish(handle trackHandle) { + if err := m.params.Transport.SendUnpublishRequest(unpublishRequestToProto(handle)); err != nil { + m.params.Logger.Debugw("could not send unpublish data track request", "handle", handle, "error", err) + } +} + +// RepublishTracks re-requests every publication after a full reconnect. Pending publications fail +// with ErrDisconnected; published tracks drop frames until the SFU answers with their new SID. +func (m *LocalManager) RepublishTracks() { + m.mu.Lock() + for handle, result := range m.pending { + delete(m.pending, handle) + result <- publishResult{err: ErrDisconnected} + } + requests := make([]publishRequest, 0, len(m.active)) + for handle, track := range m.active { + info := track.beginRepublish() + requests = append(requests, publishRequest{ + handle: handle, + name: info.Name, + usesE2EE: info.UsesE2EE, + schema: info.Schema, + frameEncoding: info.FrameEncoding, + }) + } + m.mu.Unlock() + + for _, request := range requests { + if err := m.params.Transport.SendPublishRequest(request.toProto()); err != nil { + m.params.Logger.Warnw("could not republish data track", err, "handle", request.handle) + } + } +} + +// PublishResponsesForSyncState describes the current publications for SyncState. +func (m *LocalManager) PublishResponsesForSyncState() []*livekit.PublishDataTrackResponse { + m.mu.Lock() + infos := make([]Info, 0, len(m.active)) + for _, track := range m.active { + infos = append(infos, track.Info()) + } + m.mu.Unlock() + return publishResponsesForSyncState(infos) +} + +// Shutdown fails pending publications and unpublishes every track locally. The manager can be +// used again for a new session. +func (m *LocalManager) Shutdown() { + m.mu.Lock() + for handle, result := range m.pending { + delete(m.pending, handle) + result <- publishResult{err: ErrDisconnected} + } + tracks := make([]*LocalTrack, 0, len(m.active)) + for _, track := range m.active { + tracks = append(tracks, track) + } + clear(m.active) + m.mu.Unlock() + + for _, track := range tracks { + track.markUnpublished() + } +} + +type publishState int32 + +const ( + statePublished publishState = iota + stateRepublishing + stateUnpublished +) + +// LocalTrack is a data track published by the local participant. Methods are safe for concurrent +// use. +type LocalTrack struct { + manager *LocalManager + handle trackHandle + state atomic.Int32 + unpublished chan struct{} + unpublishOnce sync.Once + + mu sync.Mutex + info Info + pipeline localPipeline +} + +func newLocalTrack(manager *LocalManager, info Info) *LocalTrack { + var encryptor Encryptor + if info.UsesE2EE { + encryptor = manager.params.Encryptor + } + return &LocalTrack{ + manager: manager, + handle: info.pubHandle, + unpublished: make(chan struct{}), + info: info, + pipeline: newLocalPipeline(info.pubHandle, encryptor), + } +} + +// Info returns a snapshot of the track's metadata. The SID changes when the local participant +// completes a full reconnect. +func (t *LocalTrack) Info() Info { + t.mu.Lock() + defer t.mu.Unlock() + return t.info +} + +// IsPublished reports whether the track is published. A track being republished after a +// reconnect still counts as published. +func (t *LocalTrack) IsPublished() bool { + return publishState(t.state.Load()) != stateUnpublished +} + +// Unpublished is closed once the track is no longer published, whether by Unpublish, the SFU, or +// disconnecting from the room. +func (t *LocalTrack) Unpublished() <-chan struct{} { + return t.unpublished +} + +// TryPush sends a frame to the track's subscribers. It returns ErrUnpublished once the track is +// unpublished and ErrQueueFull while it is being republished after a reconnect. Frames are never +// retained, so the caller may reuse the payload immediately. +func (t *LocalTrack) TryPush(frame Frame) error { + switch publishState(t.state.Load()) { + case stateRepublishing: + return ErrQueueFull + case stateUnpublished: + return ErrUnpublished + } + + t.mu.Lock() + packets, err := t.pipeline.processFrame(frame) + t.mu.Unlock() + if err != nil { + return err + } + t.manager.params.Transport.SendFrame(packets) + return nil +} + +// Unpublish stops the publication. It is safe to call more than once. +func (t *LocalTrack) Unpublish() { + t.manager.unpublish(t.handle) +} + +func (t *LocalTrack) markUnpublished() { + t.state.Store(int32(stateUnpublished)) + t.unpublishOnce.Do(func() { close(t.unpublished) }) +} + +func (t *LocalTrack) beginRepublish() Info { + t.state.Store(int32(stateRepublishing)) + return t.Info() +} + +// completeRepublish reports false if the track was not being republished. +func (t *LocalTrack) completeRepublish(sid SID) bool { + if !t.state.CompareAndSwap(int32(stateRepublishing), int32(statePublished)) { + return false + } + t.mu.Lock() + t.info.SID = sid + t.mu.Unlock() + return true +} + +// localPipeline turns frames into packets ready for the data channel. +type localPipeline struct { + encryptor Encryptor + packetizer *packetizer +} + +func newLocalPipeline(handle trackHandle, encryptor Encryptor) localPipeline { + return localPipeline{encryptor: encryptor, packetizer: newPacketizer(handle, transportMTU)} +} + +func (p *localPipeline) processFrame(frame Frame) ([][]byte, error) { + payload := frame.Payload + extensions := Extensions{UserTimestamp: frame.UserTimestamp} + if p.encryptor != nil { + ciphertext, e2ee, err := p.encryptor.Encrypt(payload) + if err != nil { + return nil, err + } + payload, extensions.E2EE = ciphertext, &e2ee + } + + packets, err := p.packetizer.packetize(payload, extensions) + if err != nil { + return nil, err + } + return marshalPackets(packets) +} + +func marshalPackets(packets []dtp.Packet) ([][]byte, error) { + raw := make([][]byte, len(packets)) + for i := range packets { + buf, err := packets[i].Marshal() + if err != nil { + return nil, err + } + raw[i] = buf + } + return raw, nil +} diff --git a/datatrack/local_test.go b/datatrack/local_test.go new file mode 100644 index 00000000..b4bd6d3e --- /dev/null +++ b/datatrack/local_test.go @@ -0,0 +1,321 @@ +// Copyright 2026 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 datatrack + +import ( + "bytes" + "context" + "testing" + "time" + + "github.com/livekit/protocol/livekit" + "github.com/livekit/protocol/utils/pointer" + "github.com/stretchr/testify/require" +) + +// fakeLocalTransport records what the manager sends, in the order it was sent. +type fakeLocalTransport struct { + publishRequests chan *livekit.PublishDataTrackRequest + unpublishRequests chan *livekit.UnpublishDataTrackRequest + frames chan [][]byte +} + +func newFakeLocalTransport() *fakeLocalTransport { + return &fakeLocalTransport{ + publishRequests: make(chan *livekit.PublishDataTrackRequest, 16), + unpublishRequests: make(chan *livekit.UnpublishDataTrackRequest, 16), + frames: make(chan [][]byte, 16), + } +} + +func (f *fakeLocalTransport) SendPublishRequest(req *livekit.PublishDataTrackRequest) error { + f.publishRequests <- req + return nil +} + +func (f *fakeLocalTransport) SendUnpublishRequest(req *livekit.UnpublishDataTrackRequest) error { + f.unpublishRequests <- req + return nil +} + +func (f *fakeLocalTransport) SendFrame(packets [][]byte) { + f.frames <- packets +} + +// prefixingEncryptor marks payloads instead of encrypting them. +type prefixingEncryptor struct{} + +func (prefixingEncryptor) Encrypt(payload []byte) ([]byte, E2EEExtension, error) { + return append([]byte{0xde, 0xad, 0xbe, 0xef}, payload...), E2EEExtension{}, nil +} + +// expectEvent receives from ch or fails after the 500 ms the Rust tests allow. +func expectEvent[T any](t *testing.T, ch <-chan T) T { + t.Helper() + select { + case event := <-ch: + return event + case <-time.After(500 * time.Millisecond): + var zero T + t.Fatalf("timed out waiting for %T", zero) + return zero + } +} + +// publishAsync calls Publish on another goroutine and delivers its outcome on the returned channel. +func publishAsync(ctx context.Context, m *LocalManager, name string) <-chan publishResult { + result := make(chan publishResult, 1) + go func() { + track, err := m.Publish(ctx, PublishOptions{Name: name}) + result <- publishResult{track: track, err: err} + }() + return result +} + +// accepted is the SFU's answer to a publish request. +func accepted(request *livekit.PublishDataTrackRequest, sid SID) *livekit.PublishDataTrackResponse { + return &livekit.PublishDataTrackResponse{Info: &livekit.DataTrackInfo{ + PubHandle: request.GetPubHandle(), + Sid: string(sid), + Name: request.GetName(), + Encryption: request.GetEncryption(), + }} +} + +func TestLocalPipeline_ProcessFrame(t *testing.T) { + pipeline := newLocalPipeline(0x8811, nil) + + const repeatedByte = 0xab + frame := Frame{ + Payload: bytes.Repeat([]byte{repeatedByte}, 32_000), + UserTimestamp: pointer.To(uint64(0x4411221111118811)), + } + + packets, err := pipeline.processFrame(frame) + require.NoError(t, err) + require.Len(t, packets, 3) + + for _, raw := range packets { + packet, err := parsePacket(raw) + require.NoError(t, err) + require.Nil(t, extensionsOf(&packet.Header).E2EE) + require.NotEmpty(t, packet.Payload) + require.Equal(t, bytes.Repeat([]byte{repeatedByte}, len(packet.Payload)), packet.Payload) + } +} + +func TestLocalManager_Shutdown(t *testing.T) { + m := NewLocalManager(LocalManagerParams{Transport: newFakeLocalTransport()}) + m.Shutdown() + require.Empty(t, m.PublishResponsesForSyncState()) +} + +func TestLocalManager_Publish(t *testing.T) { + const payloadSize, packetCount = 256, 10 + trackName, trackSID := "track", SID("DTR_1234") + + transport := newFakeLocalTransport() + m := NewLocalManager(LocalManagerParams{Transport: transport}) + + result := publishAsync(context.Background(), m, trackName) + + request := expectEvent(t, transport.publishRequests) + require.Equal(t, livekit.Encryption_NONE, request.GetEncryption()) + require.Equal(t, trackName, request.GetName()) + + // SFU accepts publication + m.HandlePublishResponse(accepted(request, trackSID)) + + res := expectEvent(t, result) + require.NoError(t, res.err) + track := res.track + require.False(t, track.Info().UsesE2EE) + require.Equal(t, trackName, track.Info().Name) + require.Equal(t, trackSID, track.Info().SID) + + for range packetCount { + require.NoError(t, track.TryPush(Frame{Payload: bytes.Repeat([]byte{0xfa}, payloadSize)})) + time.Sleep(10 * time.Millisecond) + } + for range packetCount { + packets := expectEvent(t, transport.frames) + packet, err := parsePacket(packets[0]) + require.NoError(t, err) + require.Len(t, packet.Payload, payloadSize) + } + + track.Unpublish() + unpublish := expectEvent(t, transport.unpublishRequests) + require.Equal(t, request.GetPubHandle(), unpublish.GetPubHandle()) +} + +func TestLocalManager_PublishSfuError(t *testing.T) { + transport := newFakeLocalTransport() + m := NewLocalManager(LocalManagerParams{Transport: transport}) + + result := publishAsync(context.Background(), m, "test") + + // SFU rejects publication + request := expectEvent(t, transport.publishRequests) + handled := m.HandleRequestResponse(&livekit.RequestResponse{ + Request: &livekit.RequestResponse_PublishDataTrack{PublishDataTrack: request}, + Reason: livekit.RequestResponse_LIMIT_EXCEEDED, + }) + require.True(t, handled) + + res := expectEvent(t, result) + require.ErrorIs(t, res.err, ErrLimitReached) +} + +func TestLocalManager_PublishCancelled(t *testing.T) { + transport := newFakeLocalTransport() + m := NewLocalManager(LocalManagerParams{Transport: transport}) + + ctx, cancel := context.WithCancel(context.Background()) + result := publishAsync(ctx, m, "test") + + request := expectEvent(t, transport.publishRequests) + + // Caller gives up before SFU responds + cancel() + res := expectEvent(t, result) + require.ErrorIs(t, res.err, context.Canceled) + time.Sleep(50 * time.Millisecond) + + // Late SFU response arrives after cancellation + m.HandlePublishResponse(accepted(request, "DTR_1234")) + + // Manager sends unpublish for the orphaned handle + unpublish := expectEvent(t, transport.unpublishRequests) + require.Equal(t, request.GetPubHandle(), unpublish.GetPubHandle()) +} + +func TestLocalManager_PublishWithE2EE(t *testing.T) { + transport := newFakeLocalTransport() + m := NewLocalManager(LocalManagerParams{Transport: transport, Encryptor: prefixingEncryptor{}}) + + result := publishAsync(context.Background(), m, "secure") + + // SFU publish request should indicate e2ee + request := expectEvent(t, transport.publishRequests) + require.Equal(t, livekit.Encryption_GCM, request.GetEncryption()) + + // SFU accepts publication with e2ee + m.HandlePublishResponse(accepted(request, "DTR_1234")) + + res := expectEvent(t, result) + require.NoError(t, res.err) + track := res.track + require.True(t, track.Info().UsesE2EE) + + // Push a frame and verify encryption was applied + require.NoError(t, track.TryPush(Frame{Payload: []byte{1, 2, 3, 4, 5}})) + + packets := expectEvent(t, transport.frames) + packet, err := parsePacket(packets[0]) + require.NoError(t, err) + require.Equal(t, []byte{0xde, 0xad, 0xbe, 0xef}, packet.Payload[:4]) + require.Equal(t, []byte{1, 2, 3, 4, 5}, packet.Payload[4:]) + require.NotNil(t, extensionsOf(&packet.Header).E2EE) +} + +func TestLocalManager_RepublishTracks(t *testing.T) { + transport := newFakeLocalTransport() + m := NewLocalManager(LocalManagerParams{Transport: transport}) + + // Publish a track through the full flow + trackName, trackSID := "track", SID("DTR_1234") + + result := publishAsync(context.Background(), m, trackName) + request := expectEvent(t, transport.publishRequests) + m.HandlePublishResponse(accepted(request, trackSID)) + + res := expectEvent(t, result) + require.NoError(t, res.err) + track := res.track + require.Equal(t, trackSID, track.Info().SID) + + // Simulate reconnect + m.RepublishTracks() + time.Sleep(50 * time.Millisecond) + + // TryPush should fail while republishing + require.ErrorIs(t, track.TryPush(Frame{Payload: []byte{0xff}}), ErrQueueFull) + + // SFU re-publishes with a new SID + request = expectEvent(t, transport.publishRequests) + require.Equal(t, uint32(track.handle), request.GetPubHandle()) + require.Equal(t, trackName, request.GetName()) + + newSID := SID("DTR_5678") + m.HandlePublishResponse(accepted(request, newSID)) + time.Sleep(50 * time.Millisecond) + + // SID updated in place, pushes succeed again + require.Equal(t, newSID, track.Info().SID) + require.NoError(t, track.TryPush(Frame{Payload: []byte{0xff}})) +} + +func TestLocalManager_QueryPublished(t *testing.T) { + transport := newFakeLocalTransport() + m := NewLocalManager(LocalManagerParams{Transport: transport}) + + // Publish two tracks + for _, name := range []string{"track_a", "track_b"} { + result := publishAsync(context.Background(), m, name) + request := expectEvent(t, transport.publishRequests) + m.HandlePublishResponse(accepted(request, "DTR_1234")) + require.NoError(t, expectEvent(t, result).err) + } + + published := m.PublishResponsesForSyncState() + require.Len(t, published, 2) + + var names []string + for _, response := range published { + names = append(names, response.GetInfo().GetName()) + } + require.Contains(t, names, "track_a") + require.Contains(t, names, "track_b") +} + +func TestLocalManager_ShutdownWithPendingAndActive(t *testing.T) { + transport := newFakeLocalTransport() + m := NewLocalManager(LocalManagerParams{Transport: transport}) + + // Pending publication (no SFU response sent) + pending := publishAsync(context.Background(), m, "pending") + expectEvent(t, transport.publishRequests) + + // Active publication (fully published) + active := publishAsync(context.Background(), m, "active") + request := expectEvent(t, transport.publishRequests) + m.HandlePublishResponse(accepted(request, "DTR_1234")) + + res := expectEvent(t, active) + require.NoError(t, res.err) + activeTrack := res.track + require.True(t, activeTrack.IsPublished()) + + // Shutdown the manager + m.Shutdown() + time.Sleep(50 * time.Millisecond) + + // Pending publish receives disconnected error + require.ErrorIs(t, expectEvent(t, pending).err, ErrDisconnected) + + // Active track is no longer published + require.False(t, activeTrack.IsPublished()) +} From 53b2e011a475cab3b9ab8c6501c20feff843fc62 Mon Sep 17 00:00:00 2001 From: Jacob Gelman <3182119+ladvoc@users.noreply.github.com> Date: Thu, 3 Sep 2026 11:56:42 -0700 Subject: [PATCH 03/23] Wire the local data track manager into the room --- localdatatrack.go | 37 +++++++++++++++++++++++++++++++++++++ localparticipant.go | 2 ++ room.go | 12 ++++++++++++ 3 files changed, 51 insertions(+) create mode 100644 localdatatrack.go diff --git a/localdatatrack.go b/localdatatrack.go new file mode 100644 index 00000000..d7647d4e --- /dev/null +++ b/localdatatrack.go @@ -0,0 +1,37 @@ +// Copyright 2026 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 lksdk + +import "github.com/livekit/protocol/livekit" + +// localDataTrackTransport connects the local data track manager to the engine. +type localDataTrackTransport struct { + engine *RTCEngine +} + +func (t localDataTrackTransport) SendPublishRequest(req *livekit.PublishDataTrackRequest) error { + if err := t.engine.ensurePublisherConnected(true); err != nil { + return err + } + return t.engine.SendPublishDataTrack(req) +} + +func (t localDataTrackTransport) SendUnpublishRequest(req *livekit.UnpublishDataTrackRequest) error { + return t.engine.SendUnpublishDataTrack(req) +} + +func (t localDataTrackTransport) SendFrame(packets [][]byte) { + t.engine.sendDataTrackFrame(packets) +} diff --git a/localparticipant.go b/localparticipant.go index fb651aa0..fd5cde2d 100644 --- a/localparticipant.go +++ b/localparticipant.go @@ -30,6 +30,7 @@ import ( "github.com/livekit/protocol/livekit" protoLogger "github.com/livekit/protocol/logger" + "github.com/livekit/server-sdk-go/v2/datatrack" ) const ( @@ -39,6 +40,7 @@ const ( type LocalParticipant struct { baseParticipant engine *RTCEngine + dataTracks *datatrack.LocalManager subscriptionPermission *livekit.SubscriptionPermission serverInfo *livekit.ServerInfo diff --git a/room.go b/room.go index fb0930e6..78b3b848 100644 --- a/room.go +++ b/room.go @@ -34,6 +34,7 @@ import ( protoLogger "github.com/livekit/protocol/logger" protosignalling "github.com/livekit/protocol/signalling" + "github.com/livekit/server-sdk-go/v2/datatrack" "github.com/livekit/server-sdk-go/v2/e2ee" "github.com/livekit/server-sdk-go/v2/signalling" @@ -322,6 +323,7 @@ type Room struct { sid string name string LocalParticipant *LocalParticipant + localDataTracks *datatrack.LocalManager callback *RoomCallback sidReady chan struct{} disconnectReason livekit.DisconnectReason @@ -366,6 +368,8 @@ func NewRoom(callback *RoomCallback) *Room { r.engine = NewRTCEngine(r.useSinglePeerConnection, r, r.getLocalParticipantSID, r.regionURLProvider) r.LocalParticipant = newLocalParticipant(r.engine, r.callback, r.serverInfo, r.log) + r.localDataTracks = datatrack.NewLocalManager(datatrack.LocalManagerParams{Transport: localDataTrackTransport{engine: r.engine}, Logger: r.log}) + r.LocalParticipant.dataTracks = r.localDataTracks return r } @@ -778,6 +782,7 @@ func (r *Room) sendSyncState() { Subscribe: !sendUnsub, }, PublishTracks: publishedTracks, + PublishDataTracks: r.localDataTracks.PublishResponsesForSyncState(), DataChannels: dataChannels, TrackSidsDisabled: trackSidsDisabled, // MIGRATION-TODO DatachannelReceiveStates @@ -787,6 +792,7 @@ func (r *Room) sendSyncState() { func (r *Room) cleanup() { r.engine.Close() r.LocalParticipant.closeTracks() + r.localDataTracks.Shutdown() r.setSid("", true) r.byteStreamHandlers.Clear() @@ -1106,6 +1112,7 @@ func (r *Room) OnRestarted( r.OnParticipantUpdate(otherParticipants) r.LocalParticipant.republishTracks() + r.localDataTracks.RepublishTracks() } func (r *Room) OnRestartConnected() { @@ -1401,6 +1408,9 @@ func (r *Room) OnMediaSectionsRequirement(mediaSectionsRequirement *livekit.Medi // OnRequestResponse handles responses to signal requests that no caller is waiting on. func (r *Room) OnRequestResponse(res *livekit.RequestResponse) { + if r.localDataTracks.HandleRequestResponse(res) { + return + } if res.GetReason() != livekit.RequestResponse_OK { r.log.Warnw( "signal request failed", nil, @@ -1412,9 +1422,11 @@ func (r *Room) OnRequestResponse(res *livekit.RequestResponse) { } func (r *Room) OnPublishDataTrackResponse(publishDataTrackResponse *livekit.PublishDataTrackResponse) { + r.localDataTracks.HandlePublishResponse(publishDataTrackResponse) } func (r *Room) OnUnpublishDataTrackResponse(unpublishDataTrackResponse *livekit.UnpublishDataTrackResponse) { + r.localDataTracks.HandleUnpublishResponse(unpublishDataTrackResponse) } func (r *Room) OnDataTrackSubscriberHandles(dataTrackSubscriberHandles *livekit.DataTrackSubscriberHandles) { From 0a37ffb2e3d2b6af1d1fba273d0a046963680bd9 Mon Sep 17 00:00:00 2001 From: Jacob Gelman <3182119+ladvoc@users.noreply.github.com> Date: Thu, 3 Sep 2026 11:56:42 -0700 Subject: [PATCH 04/23] Add LocalParticipant.PublishDataTrack --- localdatatrack.go | 34 +++++++++++++++++++++++++++++++++- 1 file changed, 33 insertions(+), 1 deletion(-) diff --git a/localdatatrack.go b/localdatatrack.go index d7647d4e..28c74018 100644 --- a/localdatatrack.go +++ b/localdatatrack.go @@ -14,7 +14,16 @@ package lksdk -import "github.com/livekit/protocol/livekit" +import ( + "context" + "errors" + "time" + + "github.com/livekit/protocol/livekit" + "github.com/livekit/server-sdk-go/v2/datatrack" +) + +const dataTrackPublishTimeout = 10 * time.Second // localDataTrackTransport connects the local data track manager to the engine. type localDataTrackTransport struct { @@ -35,3 +44,26 @@ func (t localDataTrackTransport) SendUnpublishRequest(req *livekit.UnpublishData func (t localDataTrackTransport) SendFrame(packets [][]byte) { t.engine.sendDataTrackFrame(packets) } + +// PublishDataTrack publishes a data track and waits for the server to accept it. A 10 second +// deadline applies when ctx has none. The name must be unique among the participant's data tracks. +func (p *LocalParticipant) PublishDataTrack(ctx context.Context, name string, opts ...datatrack.PublishOption) (*datatrack.LocalTrack, error) { + options := datatrack.PublishOptions{Name: name} + for _, opt := range opts { + opt(&options) + } + + _, hasDeadline := ctx.Deadline() + ctx, cancel := withDefaultTimeout(ctx, dataTrackPublishTimeout) + defer cancel() + + track, err := p.dataTracks.Publish(ctx, options) + if err != nil { + if !hasDeadline && errors.Is(err, context.DeadlineExceeded) { + return nil, datatrack.ErrPublishTimeout + } + return nil, err + } + p.log.Infow("published data track", "name", name, "trackID", track.Info().SID) + return track, nil +} From 455fa50282682a498800e928643e65ccbe5e7f75 Mon Sep 17 00:00:00 2001 From: Jacob Gelman <3182119+ladvoc@users.noreply.github.com> Date: Thu, 3 Sep 2026 12:30:31 -0700 Subject: [PATCH 05/23] Add data track publisher example --- examples/datatrack/publisher/main.go | 81 ++++++++++++++++++++++++++++ 1 file changed, 81 insertions(+) create mode 100644 examples/datatrack/publisher/main.go diff --git a/examples/datatrack/publisher/main.go b/examples/datatrack/publisher/main.go new file mode 100644 index 00000000..e10f770a --- /dev/null +++ b/examples/datatrack/publisher/main.go @@ -0,0 +1,81 @@ +package main + +import ( + "bytes" + "context" + "flag" + "fmt" + "os/signal" + "syscall" + "time" + + "github.com/livekit/protocol/logger" + lksdk "github.com/livekit/server-sdk-go/v2" + "github.com/livekit/server-sdk-go/v2/datatrack" +) + +var host, apiKey, apiSecret, roomName, identity string + +func init() { + flag.StringVar(&host, "host", "", "livekit server host") + flag.StringVar(&apiKey, "api-key", "", "livekit api key") + flag.StringVar(&apiSecret, "api-secret", "", "livekit api secret") + flag.StringVar(&roomName, "room-name", "", "room name") + flag.StringVar(&identity, "identity", "publisher", "participant identity") +} + +func main() { + logger.InitFromConfig(&logger.Config{Level: "info"}, "datatrack-publisher") + lksdk.SetLogger(logger.GetLogger()) + flag.Parse() + if host == "" || apiKey == "" || apiSecret == "" || roomName == "" { + fmt.Println("invalid arguments.") + return + } + + room, err := lksdk.ConnectToRoom(host, lksdk.ConnectInfo{ + APIKey: apiKey, + APISecret: apiSecret, + RoomName: roomName, + ParticipantIdentity: identity, + }, nil) + if err != nil { + panic(err) + } + defer room.Disconnect() + + track, err := room.LocalParticipant.PublishDataTrack(context.Background(), "my_sensor_data") + if err != nil { + panic(err) + } + defer track.Unpublish() + + ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) + defer stop() + pushFrames(ctx, track) +} + +func readSensor() []byte { + // Dynamically read some sensor data... + return bytes.Repeat([]byte{0xfa}, 256) +} + +func pushFrames(ctx context.Context, track *datatrack.LocalTrack) { + for { + logger.Infow("pushing frame") + + frame := datatrack.Frame{ + Payload: readSensor(), + UserTimestamp: datatrack.UserTimestampNow(), + } + if err := track.TryPush(frame); err != nil { + logger.Warnw("failed to push frame", err) + } + + select { + case <-ctx.Done(): + return + case <-time.After(500 * time.Millisecond): + } + } +} From 0f0d168ddc7de3adbbbbe75aaf9181d097312dec Mon Sep 17 00:00:00 2001 From: Jacob Gelman <3182119+ladvoc@users.noreply.github.com> Date: Mon, 14 Sep 2026 14:46:16 -0700 Subject: [PATCH 06/23] Fix handle allocator error check in Publish --- datatrack/local.go | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/datatrack/local.go b/datatrack/local.go index 346d2cd8..cd22a454 100644 --- a/datatrack/local.go +++ b/datatrack/local.go @@ -114,10 +114,10 @@ func (m *LocalManager) Publish(ctx context.Context, options PublishOptions) (*Lo } m.mu.Lock() - handle, ok := m.handles.get() - if !ok { + handle, err := m.handles.get() + if err != nil { m.mu.Unlock() - return nil, ErrLimitReached + return nil, err } result := make(chan publishResult, 1) m.pending[handle] = result From 8c07cdc109ac89f329db93b24511bcdc3d630985 Mon Sep 17 00:00:00 2001 From: Jacob Gelman <3182119+ladvoc@users.noreply.github.com> Date: Tue, 22 Sep 2026 13:41:16 -0700 Subject: [PATCH 07/23] Expose public error type for handle exhaustion --- datatrack/local.go | 2 +- datatrack/local_test.go | 9 +++++++++ 2 files changed, 10 insertions(+), 1 deletion(-) diff --git a/datatrack/local.go b/datatrack/local.go index cd22a454..e1b81c88 100644 --- a/datatrack/local.go +++ b/datatrack/local.go @@ -117,7 +117,7 @@ func (m *LocalManager) Publish(ctx context.Context, options PublishOptions) (*Lo handle, err := m.handles.get() if err != nil { m.mu.Unlock() - return nil, err + return nil, ErrLimitReached } result := make(chan publishResult, 1) m.pending[handle] = result diff --git a/datatrack/local_test.go b/datatrack/local_test.go index b4bd6d3e..ffd1add2 100644 --- a/datatrack/local_test.go +++ b/datatrack/local_test.go @@ -17,6 +17,7 @@ package datatrack import ( "bytes" "context" + "math" "testing" "time" @@ -179,6 +180,14 @@ func TestLocalManager_PublishSfuError(t *testing.T) { require.ErrorIs(t, res.err, ErrLimitReached) } +func TestLocalManager_PublishHandlesExhausted(t *testing.T) { + m := NewLocalManager(LocalManagerParams{Transport: newFakeLocalTransport()}) + m.handles.value = math.MaxUint16 + + _, err := m.Publish(context.Background(), PublishOptions{Name: "test"}) + require.ErrorIs(t, err, ErrLimitReached) +} + func TestLocalManager_PublishCancelled(t *testing.T) { transport := newFakeLocalTransport() m := NewLocalManager(LocalManagerParams{Transport: transport}) From 5a1d1224f72cd92ef99431cc1d23667818ec5a6d Mon Sep 17 00:00:00 2001 From: Jacob Gelman <3182119+ladvoc@users.noreply.github.com> Date: Tue, 22 Sep 2026 13:47:46 -0700 Subject: [PATCH 08/23] Let ensurePublisherConnected give up when its context ends --- engine.go | 17 +++++++++++------ integration_test.go | 2 +- 2 files changed, 12 insertions(+), 7 deletions(-) diff --git a/engine.go b/engine.go index d727c9e4..c3867a17 100644 --- a/engine.go +++ b/engine.go @@ -746,22 +746,24 @@ func (e *RTCEngine) GetDataChannelSub(kind livekit.DataPacket_Kind) *webrtc.Data } func (e *RTCEngine) waitUntilConnected(timeout time.Duration) error { - return waitUntilConnected(timeout, func() bool { + return waitUntilConnected(context.Background(), timeout, func() bool { return e.IsConnected() }) } -func (e *RTCEngine) ensurePublisherConnected(ensureDataReady bool) error { +// ensurePublisherConnected waits for the publisher connection, giving up when ctx ends or the +// connect timeout passes, whichever comes first. +func (e *RTCEngine) ensurePublisherConnected(ctx context.Context, ensureDataReady bool) error { e.pclock.Lock() subscriberPrimary := e.subscriberPrimary e.pclock.Unlock() connectTimeout := e.connectionManager.getConnectTimeout() if !subscriberPrimary { - return e.waitUntilConnected(connectTimeout) + return waitUntilConnected(ctx, connectTimeout, e.IsConnected) } var negotiated bool - return waitUntilConnected(connectTimeout, func() bool { + return waitUntilConnected(ctx, connectTimeout, func() bool { if publisher, ok := e.Publisher(); ok { if publisher.IsConnected() && (!ensureDataReady || e.dataPubChannelReady()) { return true @@ -1231,7 +1233,7 @@ func filterTURNServers(iceServers []*livekit.ICEServer) []*livekit.ICEServer { } func (e *RTCEngine) publishDataPacket(pck *livekit.DataPacket, kind livekit.DataPacket_Kind) error { - err := e.ensurePublisherConnected(true) + err := e.ensurePublisherConnected(context.Background(), true) if err != nil { e.log.Errorw("could not ensure publisher connected", err) return err @@ -1836,7 +1838,7 @@ func setConfiguration(pcTransport *PCTransport, configuration webrtc.Configurati } } -func waitUntilConnected(d time.Duration, test func() bool) error { +func waitUntilConnected(ctx context.Context, d time.Duration, test func() bool) error { if test() { return nil } @@ -1849,6 +1851,9 @@ func waitUntilConnected(d time.Duration, test func() bool) error { for { select { + case <-ctx.Done(): + return ctx.Err() + case <-timeout.C: return ErrConnectionTimeout diff --git a/integration_test.go b/integration_test.go index 39ad8ad5..616aba6c 100644 --- a/integration_test.go +++ b/integration_test.go @@ -833,7 +833,7 @@ func TestForceTLS(t *testing.T) { pub.LocalParticipant.PublishDataPacket(UserData([]byte("test")), WithDataPublishReliable(true)) pub.Simulate(SimulateForceTLS) - require.Eventually(t, func() bool { return reconnected.Load() && pub.engine.ensurePublisherConnected(true) == nil }, 15*time.Second, 100*time.Millisecond) + require.Eventually(t, func() bool { return reconnected.Load() && pub.engine.ensurePublisherConnected(context.Background(), true) == nil }, 15*time.Second, 100*time.Millisecond) pub.log.Infow("reconnected") From 612102066ae31f852d3513f79ae41e4bb0e37fc8 Mon Sep 17 00:00:00 2001 From: Jacob Gelman <3182119+ladvoc@users.noreply.github.com> Date: Tue, 22 Sep 2026 13:47:46 -0700 Subject: [PATCH 09/23] Pass the publish context through the data track transport --- datatrack/local.go | 7 ++++--- datatrack/local_test.go | 27 ++++++++++++++++++++++++++- localdatatrack.go | 4 ++-- localdatatrack_test.go | 38 ++++++++++++++++++++++++++++++++++++++ 4 files changed, 70 insertions(+), 6 deletions(-) create mode 100644 localdatatrack_test.go diff --git a/datatrack/local.go b/datatrack/local.go index e1b81c88..88d3a64d 100644 --- a/datatrack/local.go +++ b/datatrack/local.go @@ -30,7 +30,8 @@ const transportMTU = 16_000 // LocalTransport carries what the local manager produces: signal requests to the SFU and frame // packets over the data channel. type LocalTransport interface { - SendPublishRequest(req *livekit.PublishDataTrackRequest) error + // SendPublishRequest may block until the connection is ready; it returns ctx.Err() when ctx ends first. + SendPublishRequest(ctx context.Context, req *livekit.PublishDataTrackRequest) error SendUnpublishRequest(req *livekit.UnpublishDataTrackRequest) error // SendFrame queues all packets of one frame; it must not block. SendFrame(packets [][]byte) @@ -130,7 +131,7 @@ func (m *LocalManager) Publish(ctx context.Context, options PublishOptions) (*Lo schema: options.Schema, frameEncoding: options.FrameEncoding, } - if err := m.params.Transport.SendPublishRequest(request.toProto()); err != nil { + if err := m.params.Transport.SendPublishRequest(ctx, request.toProto()); err != nil { m.mu.Lock() delete(m.pending, handle) m.mu.Unlock() @@ -269,7 +270,7 @@ func (m *LocalManager) RepublishTracks() { m.mu.Unlock() for _, request := range requests { - if err := m.params.Transport.SendPublishRequest(request.toProto()); err != nil { + if err := m.params.Transport.SendPublishRequest(context.Background(), request.toProto()); err != nil { m.params.Logger.Warnw("could not republish data track", err, "handle", request.handle) } } diff --git a/datatrack/local_test.go b/datatrack/local_test.go index ffd1add2..57b1c96f 100644 --- a/datatrack/local_test.go +++ b/datatrack/local_test.go @@ -41,7 +41,7 @@ func newFakeLocalTransport() *fakeLocalTransport { } } -func (f *fakeLocalTransport) SendPublishRequest(req *livekit.PublishDataTrackRequest) error { +func (f *fakeLocalTransport) SendPublishRequest(_ context.Context, req *livekit.PublishDataTrackRequest) error { f.publishRequests <- req return nil } @@ -211,6 +211,31 @@ func TestLocalManager_PublishCancelled(t *testing.T) { require.Equal(t, request.GetPubHandle(), unpublish.GetPubHandle()) } +// blockingTransport never becomes ready: SendPublishRequest waits on ctx like the real transport +// waits on the engine connection. +type blockingTransport struct{ *fakeLocalTransport } + +func (b blockingTransport) SendPublishRequest(ctx context.Context, _ *livekit.PublishDataTrackRequest) error { + <-ctx.Done() + return ctx.Err() +} + +func TestLocalManager_PublishCancelledWhileTransportBlocked(t *testing.T) { + transport := newFakeLocalTransport() + m := NewLocalManager(LocalManagerParams{Transport: blockingTransport{transport}}) + + ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer cancel() + res := expectEvent(t, publishAsync(ctx, m, "test")) + require.ErrorIs(t, res.err, context.DeadlineExceeded) + + // The abandoned request released its handle and left nothing pending + m.mu.Lock() + defer m.mu.Unlock() + require.Empty(t, m.pending) + require.Empty(t, m.active) +} + func TestLocalManager_PublishWithE2EE(t *testing.T) { transport := newFakeLocalTransport() m := NewLocalManager(LocalManagerParams{Transport: transport, Encryptor: prefixingEncryptor{}}) diff --git a/localdatatrack.go b/localdatatrack.go index 28c74018..75ea19be 100644 --- a/localdatatrack.go +++ b/localdatatrack.go @@ -30,8 +30,8 @@ type localDataTrackTransport struct { engine *RTCEngine } -func (t localDataTrackTransport) SendPublishRequest(req *livekit.PublishDataTrackRequest) error { - if err := t.engine.ensurePublisherConnected(true); err != nil { +func (t localDataTrackTransport) SendPublishRequest(ctx context.Context, req *livekit.PublishDataTrackRequest) error { + if err := t.engine.ensurePublisherConnected(ctx, true); err != nil { return err } return t.engine.SendPublishDataTrack(req) diff --git a/localdatatrack_test.go b/localdatatrack_test.go new file mode 100644 index 00000000..60e4ebb3 --- /dev/null +++ b/localdatatrack_test.go @@ -0,0 +1,38 @@ +// Copyright 2026 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 lksdk + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +// The room never connects, so the publish request waits on the engine. The caller's deadline must +// end that wait rather than the 15 s connect timeout. +func TestPublishDataTrackDeadlineInterruptsConnectionWait(t *testing.T) { + room := NewRoom(&RoomCallback{}) + t.Cleanup(room.engine.Close) + + ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer cancel() + + start := time.Now() + _, err := room.LocalParticipant.PublishDataTrack(ctx, "test") + require.ErrorIs(t, err, context.DeadlineExceeded) + require.Less(t, time.Since(start), time.Second) +} From 5e1892753b25e514ba30eb2838b20e14bca7a380 Mon Sep 17 00:00:00 2001 From: Jacob Gelman <3182119+ladvoc@users.noreply.github.com> Date: Mon, 19 Oct 2026 10:06:19 -0700 Subject: [PATCH 10/23] Drop references to Rust --- datatrack/local_test.go | 3 ++- datatrack/packet_test.go | 4 ++-- 2 files changed, 4 insertions(+), 3 deletions(-) diff --git a/datatrack/local_test.go b/datatrack/local_test.go index 57b1c96f..2595c317 100644 --- a/datatrack/local_test.go +++ b/datatrack/local_test.go @@ -62,7 +62,8 @@ func (prefixingEncryptor) Encrypt(payload []byte) ([]byte, E2EEExtension, error) return append([]byte{0xde, 0xad, 0xbe, 0xef}, payload...), E2EEExtension{}, nil } -// expectEvent receives from ch or fails after the 500 ms the Rust tests allow. +// expectEvent receives from ch or fails fast. The 500 ms timeout is generous +// enough to avoid flakes on slow CI while keeping failures quick. func expectEvent[T any](t *testing.T, ch <-chan T) T { t.Helper() select { diff --git a/datatrack/packet_test.go b/datatrack/packet_test.go index aa120dcf..5fca6c06 100644 --- a/datatrack/packet_test.go +++ b/datatrack/packet_test.go @@ -25,7 +25,7 @@ import ( "github.com/stretchr/testify/require" ) -// testHeader is the header of the packet used by the Rust serialization tests, without extensions. +// testHeader is a representative packet header without extensions. func testHeader() dtp.Header { header := dtp.Header{ Version: supportedVersion, @@ -49,7 +49,7 @@ func testExtensions() Extensions { } } -// testPacket is the packet used by the Rust serialization tests. +// testPacket is a representative packet with all extensions. func testPacket() dtp.Packet { header := testHeader() if err := testExtensions().apply(&header); err != nil { From 10843005265f8c36310c18b94a6badc263cfd1fe Mon Sep 17 00:00:00 2001 From: Jacob Gelman <3182119+ladvoc@users.noreply.github.com> Date: Mon, 19 Oct 2026 10:45:13 -0700 Subject: [PATCH 11/23] Use cli package for example --- examples/datatrack/publisher/main.go | 57 ++++++++++++++++------------ go.mod | 1 + go.sum | 2 + 3 files changed, 35 insertions(+), 25 deletions(-) diff --git a/examples/datatrack/publisher/main.go b/examples/datatrack/publisher/main.go index e10f770a..2586e0b7 100644 --- a/examples/datatrack/publisher/main.go +++ b/examples/datatrack/publisher/main.go @@ -3,8 +3,8 @@ package main import ( "bytes" "context" - "flag" "fmt" + "os" "os/signal" "syscall" "time" @@ -12,47 +12,54 @@ import ( "github.com/livekit/protocol/logger" lksdk "github.com/livekit/server-sdk-go/v2" "github.com/livekit/server-sdk-go/v2/datatrack" + "github.com/urfave/cli/v3" ) -var host, apiKey, apiSecret, roomName, identity string - -func init() { - flag.StringVar(&host, "host", "", "livekit server host") - flag.StringVar(&apiKey, "api-key", "", "livekit api key") - flag.StringVar(&apiSecret, "api-secret", "", "livekit api secret") - flag.StringVar(&roomName, "room-name", "", "room name") - flag.StringVar(&identity, "identity", "publisher", "participant identity") -} - func main() { logger.InitFromConfig(&logger.Config{Level: "info"}, "datatrack-publisher") lksdk.SetLogger(logger.GetLogger()) - flag.Parse() - if host == "" || apiKey == "" || apiSecret == "" || roomName == "" { - fmt.Println("invalid arguments.") - return + + cmd := &cli.Command{ + Name: "datatrack-publisher", + Usage: "publish sensor data on a data track", + Flags: []cli.Flag{ + &cli.StringFlag{Name: "host", Usage: "livekit server host", Required: true, Sources: cli.EnvVars("LIVEKIT_URL")}, + &cli.StringFlag{Name: "api-key", Usage: "livekit api key", Required: true, Sources: cli.EnvVars("LIVEKIT_API_KEY")}, + &cli.StringFlag{Name: "api-secret", Usage: "livekit api secret", Required: true, Sources: cli.EnvVars("LIVEKIT_API_SECRET")}, + &cli.StringFlag{Name: "room-name", Usage: "room name", Required: true}, + &cli.StringFlag{Name: "identity", Usage: "participant identity", Value: "publisher"}, + }, + Action: run, } - room, err := lksdk.ConnectToRoom(host, lksdk.ConnectInfo{ - APIKey: apiKey, - APISecret: apiSecret, - RoomName: roomName, - ParticipantIdentity: identity, + ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) + defer stop() + if err := cmd.Run(ctx, os.Args); err != nil { + fmt.Fprintln(os.Stderr, err) + os.Exit(1) + } +} + +func run(ctx context.Context, cmd *cli.Command) error { + room, err := lksdk.ConnectToRoom(cmd.String("host"), lksdk.ConnectInfo{ + APIKey: cmd.String("api-key"), + APISecret: cmd.String("api-secret"), + RoomName: cmd.String("room-name"), + ParticipantIdentity: cmd.String("identity"), }, nil) if err != nil { - panic(err) + return err } defer room.Disconnect() - track, err := room.LocalParticipant.PublishDataTrack(context.Background(), "my_sensor_data") + track, err := room.LocalParticipant.PublishDataTrack(ctx, "my_sensor_data") if err != nil { - panic(err) + return err } defer track.Unpublish() - ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) - defer stop() pushFrames(ctx, track) + return nil } func readSensor() []byte { diff --git a/go.mod b/go.mod index ec8bc220..45a79eb8 100644 --- a/go.mod +++ b/go.mod @@ -22,6 +22,7 @@ require ( github.com/pion/webrtc/v4 v4.2.24 github.com/stretchr/testify v1.12.1 github.com/twitchtv/twirp v8.1.3+incompatible + github.com/urfave/cli/v3 v3.9.0 go.uber.org/atomic v1.12.0 golang.org/x/crypto v0.57.0 google.golang.org/protobuf v1.36.12 diff --git a/go.sum b/go.sum index 3e3fe4b1..0f599f59 100644 --- a/go.sum +++ b/go.sum @@ -282,6 +282,8 @@ github.com/tonistiigi/vt100 v0.0.0-20240514184818-90bafcd6abab h1:H6aJ0yKQ0gF49Q github.com/tonistiigi/vt100 v0.0.0-20240514184818-90bafcd6abab/go.mod h1:ulncasL3N9uLrVann0m+CDlJKWsIAP34MPcOJF6VRvc= github.com/twitchtv/twirp v8.1.3+incompatible h1:+F4TdErPgSUbMZMwp13Q/KgDVuI7HJXP61mNV3/7iuU= github.com/twitchtv/twirp v8.1.3+incompatible/go.mod h1:RRJoFSAmTEh2weEqWtpPE3vFK5YBhA6bqp2l1kfCC5A= +github.com/urfave/cli/v3 v3.9.0 h1:AV9lIiPv3ukYnxunaCUsHnEozptYmDN2F0+yWqLMn/c= +github.com/urfave/cli/v3 v3.9.0/go.mod h1:ysVLtOEmg2tOy6PknnYVhDoouyC/6N42TMeoMzskhso= github.com/vbatts/tar-split v0.12.3 h1:Cd46rkGXI3Td4yrVNwU8ripbxFaQbmesqhjBUUYAJSw= github.com/vbatts/tar-split v0.12.3/go.mod h1:sQOc6OlqGCr7HkGx/IDBeKiTIvqhmj8KffNhEXG4Nq0= github.com/wlynxg/anet v0.0.5 h1:J3VJGi1gvo0JwZ/P1/Yc/8p63SoW98B5dHkYDmpgvvU= From 7ffff3e2586bfa0ce061b8099a72408381222d09 Mon Sep 17 00:00:00 2001 From: Jacob Gelman <3182119+ladvoc@users.noreply.github.com> Date: Mon, 19 Oct 2026 10:47:33 -0700 Subject: [PATCH 12/23] Use upstream E2EE extension and TrackID types in local track code --- datatrack/local.go | 4 ++-- datatrack/local_test.go | 13 +++++++------ 2 files changed, 9 insertions(+), 8 deletions(-) diff --git a/datatrack/local.go b/datatrack/local.go index 88d3a64d..47fa91d1 100644 --- a/datatrack/local.go +++ b/datatrack/local.go @@ -39,7 +39,7 @@ type LocalTransport interface { // Encryptor seals frame payloads for end-to-end encryption. type Encryptor interface { - Encrypt(payload []byte) ([]byte, E2EEExtension, error) + Encrypt(payload []byte) ([]byte, dtp.ExtensionE2EE, error) } // PublishOptions describe a data track to publish. @@ -400,7 +400,7 @@ func (t *LocalTrack) beginRepublish() Info { } // completeRepublish reports false if the track was not being republished. -func (t *LocalTrack) completeRepublish(sid SID) bool { +func (t *LocalTrack) completeRepublish(sid livekit.TrackID) bool { if !t.state.CompareAndSwap(int32(stateRepublishing), int32(statePublished)) { return false } diff --git a/datatrack/local_test.go b/datatrack/local_test.go index 2595c317..5603b6e8 100644 --- a/datatrack/local_test.go +++ b/datatrack/local_test.go @@ -21,6 +21,7 @@ import ( "testing" "time" + dtp "github.com/livekit/protocol/datatrack" "github.com/livekit/protocol/livekit" "github.com/livekit/protocol/utils/pointer" "github.com/stretchr/testify/require" @@ -58,8 +59,8 @@ func (f *fakeLocalTransport) SendFrame(packets [][]byte) { // prefixingEncryptor marks payloads instead of encrypting them. type prefixingEncryptor struct{} -func (prefixingEncryptor) Encrypt(payload []byte) ([]byte, E2EEExtension, error) { - return append([]byte{0xde, 0xad, 0xbe, 0xef}, payload...), E2EEExtension{}, nil +func (prefixingEncryptor) Encrypt(payload []byte) ([]byte, dtp.ExtensionE2EE, error) { + return append([]byte{0xde, 0xad, 0xbe, 0xef}, payload...), dtp.ExtensionE2EE{}, nil } // expectEvent receives from ch or fails fast. The 500 ms timeout is generous @@ -87,7 +88,7 @@ func publishAsync(ctx context.Context, m *LocalManager, name string) <-chan publ } // accepted is the SFU's answer to a publish request. -func accepted(request *livekit.PublishDataTrackRequest, sid SID) *livekit.PublishDataTrackResponse { +func accepted(request *livekit.PublishDataTrackRequest, sid livekit.TrackID) *livekit.PublishDataTrackResponse { return &livekit.PublishDataTrackResponse{Info: &livekit.DataTrackInfo{ PubHandle: request.GetPubHandle(), Sid: string(sid), @@ -126,7 +127,7 @@ func TestLocalManager_Shutdown(t *testing.T) { func TestLocalManager_Publish(t *testing.T) { const payloadSize, packetCount = 256, 10 - trackName, trackSID := "track", SID("DTR_1234") + trackName, trackSID := "track", livekit.TrackID("DTR_1234") transport := newFakeLocalTransport() m := NewLocalManager(LocalManagerParams{Transport: transport}) @@ -271,7 +272,7 @@ func TestLocalManager_RepublishTracks(t *testing.T) { m := NewLocalManager(LocalManagerParams{Transport: transport}) // Publish a track through the full flow - trackName, trackSID := "track", SID("DTR_1234") + trackName, trackSID := "track", livekit.TrackID("DTR_1234") result := publishAsync(context.Background(), m, trackName) request := expectEvent(t, transport.publishRequests) @@ -294,7 +295,7 @@ func TestLocalManager_RepublishTracks(t *testing.T) { require.Equal(t, uint32(track.handle), request.GetPubHandle()) require.Equal(t, trackName, request.GetName()) - newSID := SID("DTR_5678") + newSID := livekit.TrackID("DTR_5678") m.HandlePublishResponse(accepted(request, newSID)) time.Sleep(50 * time.Millisecond) From 859d92aba46dea7bd2361ecef3c517b6166fe601 Mon Sep 17 00:00:00 2001 From: Jacob Gelman <3182119+ladvoc@users.noreply.github.com> Date: Mon, 19 Oct 2026 10:52:16 -0700 Subject: [PATCH 13/23] Move data track implmentation to internal package --- datatrack/datatrack.go | 108 ++++++++++++++++++ {datatrack => internal/datatrack}/clock.go | 0 .../datatrack}/depacketizer.go | 0 .../datatrack}/depacketizer_test.go | 0 {datatrack => internal/datatrack}/errors.go | 0 {datatrack => internal/datatrack}/frame.go | 0 {datatrack => internal/datatrack}/handle.go | 0 {datatrack => internal/datatrack}/info.go | 0 {datatrack => internal/datatrack}/local.go | 0 .../datatrack}/local_test.go | 0 {datatrack => internal/datatrack}/packet.go | 0 .../datatrack}/packet_test.go | 0 .../datatrack}/packetizer.go | 0 .../datatrack}/packetizer_test.go | 0 {datatrack => internal/datatrack}/proto.go | 0 .../datatrack}/proto_test.go | 0 {datatrack => internal/datatrack}/schema.go | 0 .../datatrack}/schema_test.go | 0 localparticipant.go | 4 +- room.go | 6 +- 20 files changed, 113 insertions(+), 5 deletions(-) create mode 100644 datatrack/datatrack.go rename {datatrack => internal/datatrack}/clock.go (100%) rename {datatrack => internal/datatrack}/depacketizer.go (100%) rename {datatrack => internal/datatrack}/depacketizer_test.go (100%) rename {datatrack => internal/datatrack}/errors.go (100%) rename {datatrack => internal/datatrack}/frame.go (100%) rename {datatrack => internal/datatrack}/handle.go (100%) rename {datatrack => internal/datatrack}/info.go (100%) rename {datatrack => internal/datatrack}/local.go (100%) rename {datatrack => internal/datatrack}/local_test.go (100%) rename {datatrack => internal/datatrack}/packet.go (100%) rename {datatrack => internal/datatrack}/packet_test.go (100%) rename {datatrack => internal/datatrack}/packetizer.go (100%) rename {datatrack => internal/datatrack}/packetizer_test.go (100%) rename {datatrack => internal/datatrack}/proto.go (100%) rename {datatrack => internal/datatrack}/proto_test.go (100%) rename {datatrack => internal/datatrack}/schema.go (100%) rename {datatrack => internal/datatrack}/schema_test.go (100%) diff --git a/datatrack/datatrack.go b/datatrack/datatrack.go new file mode 100644 index 00000000..bf38b992 --- /dev/null +++ b/datatrack/datatrack.go @@ -0,0 +1,108 @@ +// Copyright 2026 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 datatrack is the public API for data tracks. Publish a track with +// [github.com/livekit/server-sdk-go/v2.LocalParticipant.PublishDataTrack]. +package datatrack + +import ( + backend "github.com/livekit/server-sdk-go/v2/internal/datatrack" +) + +type ( + // LocalTrack is a data track published by the local participant. + LocalTrack = backend.LocalTrack + // Info describes a published data track. + Info = backend.Info + // Frame is a frame published on a data track. + Frame = backend.Frame + + // PublishOptions describe a data track to publish. + PublishOptions = backend.PublishOptions + // PublishOption customizes a publication. + PublishOption = backend.PublishOption + + // SchemaID identifies a data track schema. + SchemaID = backend.SchemaID + // SchemaEncoding is the encoding of a schema definition: a WellKnownSchemaEncoding or a + // CustomSchemaEncoding. + SchemaEncoding = backend.SchemaEncoding + // WellKnownSchemaEncoding is a schema encoding defined by the protocol. + WellKnownSchemaEncoding = backend.WellKnownSchemaEncoding + // CustomSchemaEncoding is an application-defined schema encoding identified by name. + CustomSchemaEncoding = backend.CustomSchemaEncoding + + // FrameEncoding is the encoding of frames pushed on a data track: a WellKnownFrameEncoding or + // a CustomFrameEncoding. + FrameEncoding = backend.FrameEncoding + // WellKnownFrameEncoding is a frame encoding defined by the protocol. + WellKnownFrameEncoding = backend.WellKnownFrameEncoding + // CustomFrameEncoding is an application-defined frame encoding identified by name. + CustomFrameEncoding = backend.CustomFrameEncoding +) + +var ( + ErrNotAllowed = backend.ErrNotAllowed + ErrDuplicateName = backend.ErrDuplicateName + ErrInvalidName = backend.ErrInvalidName + ErrLimitReached = backend.ErrLimitReached + ErrPublishTimeout = backend.ErrPublishTimeout + ErrDisconnected = backend.ErrDisconnected + ErrUnpublished = backend.ErrUnpublished + ErrQueueFull = backend.ErrQueueFull + + ErrSchemaMissingFrameEncoding = backend.ErrSchemaMissingFrameEncoding + ErrSchemaMissingID = backend.ErrSchemaMissingID + ErrSchemaIncompatible = backend.ErrSchemaIncompatible +) + +const ( + // SchemaEncodingOther is a well-known encoding not known to this SDK version. + SchemaEncodingOther = backend.SchemaEncodingOther + SchemaEncodingProtobuf = backend.SchemaEncodingProtobuf + SchemaEncodingFlatbuffer = backend.SchemaEncodingFlatbuffer + SchemaEncodingROS1Msg = backend.SchemaEncodingROS1Msg + SchemaEncodingROS2Msg = backend.SchemaEncodingROS2Msg + SchemaEncodingROS2IDL = backend.SchemaEncodingROS2IDL + SchemaEncodingOMGIDL = backend.SchemaEncodingOMGIDL + SchemaEncodingJSONSchema = backend.SchemaEncodingJSONSchema +) + +const ( + // FrameEncodingOther is a well-known encoding not known to this SDK version. + FrameEncodingOther = backend.FrameEncodingOther + FrameEncodingROS1 = backend.FrameEncodingROS1 + FrameEncodingCDR = backend.FrameEncodingCDR + FrameEncodingProtobuf = backend.FrameEncodingProtobuf + FrameEncodingFlatbuffer = backend.FrameEncodingFlatbuffer + FrameEncodingCBOR = backend.FrameEncodingCBOR + FrameEncodingMsgPack = backend.FrameEncodingMsgPack + FrameEncodingJSON = backend.FrameEncodingJSON +) + +// WithSchema associates a schema with the track's frames. +func WithSchema(schema SchemaID) PublishOption { + return backend.WithSchema(schema) +} + +// WithFrameEncoding declares the encoding of the track's frames. +func WithFrameEncoding(encoding FrameEncoding) PublishOption { + return backend.WithFrameEncoding(encoding) +} + +// UserTimestampNow returns the current Unix time in milliseconds, the interpretation +// Frame.DurationSinceTimestamp expects. +func UserTimestampNow() *uint64 { + return backend.UserTimestampNow() +} diff --git a/datatrack/clock.go b/internal/datatrack/clock.go similarity index 100% rename from datatrack/clock.go rename to internal/datatrack/clock.go diff --git a/datatrack/depacketizer.go b/internal/datatrack/depacketizer.go similarity index 100% rename from datatrack/depacketizer.go rename to internal/datatrack/depacketizer.go diff --git a/datatrack/depacketizer_test.go b/internal/datatrack/depacketizer_test.go similarity index 100% rename from datatrack/depacketizer_test.go rename to internal/datatrack/depacketizer_test.go diff --git a/datatrack/errors.go b/internal/datatrack/errors.go similarity index 100% rename from datatrack/errors.go rename to internal/datatrack/errors.go diff --git a/datatrack/frame.go b/internal/datatrack/frame.go similarity index 100% rename from datatrack/frame.go rename to internal/datatrack/frame.go diff --git a/datatrack/handle.go b/internal/datatrack/handle.go similarity index 100% rename from datatrack/handle.go rename to internal/datatrack/handle.go diff --git a/datatrack/info.go b/internal/datatrack/info.go similarity index 100% rename from datatrack/info.go rename to internal/datatrack/info.go diff --git a/datatrack/local.go b/internal/datatrack/local.go similarity index 100% rename from datatrack/local.go rename to internal/datatrack/local.go diff --git a/datatrack/local_test.go b/internal/datatrack/local_test.go similarity index 100% rename from datatrack/local_test.go rename to internal/datatrack/local_test.go diff --git a/datatrack/packet.go b/internal/datatrack/packet.go similarity index 100% rename from datatrack/packet.go rename to internal/datatrack/packet.go diff --git a/datatrack/packet_test.go b/internal/datatrack/packet_test.go similarity index 100% rename from datatrack/packet_test.go rename to internal/datatrack/packet_test.go diff --git a/datatrack/packetizer.go b/internal/datatrack/packetizer.go similarity index 100% rename from datatrack/packetizer.go rename to internal/datatrack/packetizer.go diff --git a/datatrack/packetizer_test.go b/internal/datatrack/packetizer_test.go similarity index 100% rename from datatrack/packetizer_test.go rename to internal/datatrack/packetizer_test.go diff --git a/datatrack/proto.go b/internal/datatrack/proto.go similarity index 100% rename from datatrack/proto.go rename to internal/datatrack/proto.go diff --git a/datatrack/proto_test.go b/internal/datatrack/proto_test.go similarity index 100% rename from datatrack/proto_test.go rename to internal/datatrack/proto_test.go diff --git a/datatrack/schema.go b/internal/datatrack/schema.go similarity index 100% rename from datatrack/schema.go rename to internal/datatrack/schema.go diff --git a/datatrack/schema_test.go b/internal/datatrack/schema_test.go similarity index 100% rename from datatrack/schema_test.go rename to internal/datatrack/schema_test.go diff --git a/localparticipant.go b/localparticipant.go index fd5cde2d..1cc8b6b3 100644 --- a/localparticipant.go +++ b/localparticipant.go @@ -30,7 +30,7 @@ import ( "github.com/livekit/protocol/livekit" protoLogger "github.com/livekit/protocol/logger" - "github.com/livekit/server-sdk-go/v2/datatrack" + dtbackend "github.com/livekit/server-sdk-go/v2/internal/datatrack" ) const ( @@ -40,7 +40,7 @@ const ( type LocalParticipant struct { baseParticipant engine *RTCEngine - dataTracks *datatrack.LocalManager + dataTracks *dtbackend.LocalManager subscriptionPermission *livekit.SubscriptionPermission serverInfo *livekit.ServerInfo diff --git a/room.go b/room.go index 78b3b848..0fb18e51 100644 --- a/room.go +++ b/room.go @@ -34,8 +34,8 @@ import ( protoLogger "github.com/livekit/protocol/logger" protosignalling "github.com/livekit/protocol/signalling" - "github.com/livekit/server-sdk-go/v2/datatrack" "github.com/livekit/server-sdk-go/v2/e2ee" + dtbackend "github.com/livekit/server-sdk-go/v2/internal/datatrack" "github.com/livekit/server-sdk-go/v2/signalling" "github.com/livekit/mediatransportutil/pkg/pacer" @@ -323,7 +323,7 @@ type Room struct { sid string name string LocalParticipant *LocalParticipant - localDataTracks *datatrack.LocalManager + localDataTracks *dtbackend.LocalManager callback *RoomCallback sidReady chan struct{} disconnectReason livekit.DisconnectReason @@ -368,7 +368,7 @@ func NewRoom(callback *RoomCallback) *Room { r.engine = NewRTCEngine(r.useSinglePeerConnection, r, r.getLocalParticipantSID, r.regionURLProvider) r.LocalParticipant = newLocalParticipant(r.engine, r.callback, r.serverInfo, r.log) - r.localDataTracks = datatrack.NewLocalManager(datatrack.LocalManagerParams{Transport: localDataTrackTransport{engine: r.engine}, Logger: r.log}) + r.localDataTracks = dtbackend.NewLocalManager(dtbackend.LocalManagerParams{Transport: localDataTrackTransport{engine: r.engine}, Logger: r.log}) r.LocalParticipant.dataTracks = r.localDataTracks return r } From 50338cf1657ffe1015416ea279a87f11a7c29610 Mon Sep 17 00:00:00 2001 From: Jacob Gelman <3182119+ladvoc@users.noreply.github.com> Date: Mon, 19 Oct 2026 11:00:30 -0700 Subject: [PATCH 14/23] Use type alias for frame packets --- datatracksender.go | 15 +++++++-------- datatracksender_test.go | 6 ++++-- engine.go | 3 ++- internal/datatrack/local.go | 13 ++++++++----- internal/datatrack/local_test.go | 8 ++++---- localdatatrack.go | 5 +++-- 6 files changed, 28 insertions(+), 22 deletions(-) diff --git a/datatracksender.go b/datatracksender.go index 495a0153..1c95dacc 100644 --- a/datatracksender.go +++ b/datatracksender.go @@ -19,13 +19,12 @@ import ( protoLogger "github.com/livekit/protocol/logger" "github.com/pion/webrtc/v4" + + dtbackend "github.com/livekit/server-sdk-go/v2/internal/datatrack" ) const dataTrackBufferedAmountLowThreshold = 8 * 1024 -// dataTrackFramePackets is one frame serialized into data track packets. -type dataTrackFramePackets [][]byte - // dataTrackSender paces frames onto the data track channel, keeping only the freshest frame // while the channel's send buffer is above the low threshold. type dataTrackSender struct { @@ -33,7 +32,7 @@ type dataTrackSender struct { lock sync.Mutex dc *webrtc.DataChannel - frame dataTrackFramePackets + frame dtbackend.FramePackets notify chan struct{} done chan struct{} } @@ -61,7 +60,7 @@ func (s *dataTrackSender) setLogger(log protoLogger.Logger) { s.log = log } -func (s *dataTrackSender) send(frame dataTrackFramePackets) { +func (s *dataTrackSender) send(frame dtbackend.FramePackets) { if dropped := s.push(frame); dropped != nil { s.log.Debugw("dropping data track frame", "numPackets", len(dropped)) } @@ -78,7 +77,7 @@ func (s *dataTrackSender) stop() { close(s.done) } -func (s *dataTrackSender) push(frame dataTrackFramePackets) (dropped dataTrackFramePackets) { +func (s *dataTrackSender) push(frame dtbackend.FramePackets) (dropped dtbackend.FramePackets) { if len(frame) == 0 { return nil } @@ -91,7 +90,7 @@ func (s *dataTrackSender) push(frame dataTrackFramePackets) (dropped dataTrackFr return dropped } -func (s *dataTrackSender) pop() dataTrackFramePackets { +func (s *dataTrackSender) pop() dtbackend.FramePackets { s.lock.Lock() defer s.lock.Unlock() @@ -103,7 +102,7 @@ func (s *dataTrackSender) pop() dataTrackFramePackets { func (s *dataTrackSender) run() { var ( dc *webrtc.DataChannel - inFlight dataTrackFramePackets + inFlight dtbackend.FramePackets ) for { select { diff --git a/datatracksender_test.go b/datatracksender_test.go index a76ae0e3..a572e86d 100644 --- a/datatracksender_test.go +++ b/datatracksender_test.go @@ -18,10 +18,12 @@ import ( "testing" "github.com/stretchr/testify/require" + + dtbackend "github.com/livekit/server-sdk-go/v2/internal/datatrack" ) -func testFrame(marker byte, packets int) dataTrackFramePackets { - frame := make(dataTrackFramePackets, packets) +func testFrame(marker byte, packets int) dtbackend.FramePackets { + frame := make(dtbackend.FramePackets, packets) for i := range frame { frame[i] = []byte{marker, byte(i)} } diff --git a/engine.go b/engine.go index c3867a17..1bd5415d 100644 --- a/engine.go +++ b/engine.go @@ -34,6 +34,7 @@ import ( protosignalling "github.com/livekit/protocol/signalling" "github.com/livekit/server-sdk-go/v2/e2ee" + dtbackend "github.com/livekit/server-sdk-go/v2/internal/datatrack" "github.com/livekit/server-sdk-go/v2/signalling" ) @@ -1865,6 +1866,6 @@ func waitUntilConnected(ctx context.Context, d time.Duration, test func() bool) } } -func (e *RTCEngine) sendDataTrackFrame(frame dataTrackFramePackets) { +func (e *RTCEngine) sendDataTrackFrame(frame dtbackend.FramePackets) { e.dataTrackSender.send(frame) } diff --git a/internal/datatrack/local.go b/internal/datatrack/local.go index 47fa91d1..11c28bc2 100644 --- a/internal/datatrack/local.go +++ b/internal/datatrack/local.go @@ -27,14 +27,17 @@ import ( // transportMTU is the largest packet sent over the data channel. const transportMTU = 16_000 +// FramePackets is one frame serialized into data track packets. +type FramePackets [][]byte + // LocalTransport carries what the local manager produces: signal requests to the SFU and frame // packets over the data channel. type LocalTransport interface { // SendPublishRequest may block until the connection is ready; it returns ctx.Err() when ctx ends first. SendPublishRequest(ctx context.Context, req *livekit.PublishDataTrackRequest) error SendUnpublishRequest(req *livekit.UnpublishDataTrackRequest) error - // SendFrame queues all packets of one frame; it must not block. - SendFrame(packets [][]byte) + // SendFrame queues a frame's packets; it must not block. + SendFrame(frame FramePackets) } // Encryptor seals frame payloads for end-to-end encryption. @@ -420,7 +423,7 @@ func newLocalPipeline(handle trackHandle, encryptor Encryptor) localPipeline { return localPipeline{encryptor: encryptor, packetizer: newPacketizer(handle, transportMTU)} } -func (p *localPipeline) processFrame(frame Frame) ([][]byte, error) { +func (p *localPipeline) processFrame(frame Frame) (FramePackets, error) { payload := frame.Payload extensions := Extensions{UserTimestamp: frame.UserTimestamp} if p.encryptor != nil { @@ -438,8 +441,8 @@ func (p *localPipeline) processFrame(frame Frame) ([][]byte, error) { return marshalPackets(packets) } -func marshalPackets(packets []dtp.Packet) ([][]byte, error) { - raw := make([][]byte, len(packets)) +func marshalPackets(packets []dtp.Packet) (FramePackets, error) { + raw := make(FramePackets, len(packets)) for i := range packets { buf, err := packets[i].Marshal() if err != nil { diff --git a/internal/datatrack/local_test.go b/internal/datatrack/local_test.go index 5603b6e8..a7fffdf4 100644 --- a/internal/datatrack/local_test.go +++ b/internal/datatrack/local_test.go @@ -31,14 +31,14 @@ import ( type fakeLocalTransport struct { publishRequests chan *livekit.PublishDataTrackRequest unpublishRequests chan *livekit.UnpublishDataTrackRequest - frames chan [][]byte + frames chan FramePackets } func newFakeLocalTransport() *fakeLocalTransport { return &fakeLocalTransport{ publishRequests: make(chan *livekit.PublishDataTrackRequest, 16), unpublishRequests: make(chan *livekit.UnpublishDataTrackRequest, 16), - frames: make(chan [][]byte, 16), + frames: make(chan FramePackets, 16), } } @@ -52,8 +52,8 @@ func (f *fakeLocalTransport) SendUnpublishRequest(req *livekit.UnpublishDataTrac return nil } -func (f *fakeLocalTransport) SendFrame(packets [][]byte) { - f.frames <- packets +func (f *fakeLocalTransport) SendFrame(frame FramePackets) { + f.frames <- frame } // prefixingEncryptor marks payloads instead of encrypting them. diff --git a/localdatatrack.go b/localdatatrack.go index 75ea19be..2717951a 100644 --- a/localdatatrack.go +++ b/localdatatrack.go @@ -21,6 +21,7 @@ import ( "github.com/livekit/protocol/livekit" "github.com/livekit/server-sdk-go/v2/datatrack" + dtbackend "github.com/livekit/server-sdk-go/v2/internal/datatrack" ) const dataTrackPublishTimeout = 10 * time.Second @@ -41,8 +42,8 @@ func (t localDataTrackTransport) SendUnpublishRequest(req *livekit.UnpublishData return t.engine.SendUnpublishDataTrack(req) } -func (t localDataTrackTransport) SendFrame(packets [][]byte) { - t.engine.sendDataTrackFrame(packets) +func (t localDataTrackTransport) SendFrame(frame dtbackend.FramePackets) { + t.engine.sendDataTrackFrame(frame) } // PublishDataTrack publishes a data track and waits for the server to accept it. A 10 second From ae3e8bdb68b6e939624c2d85d8b7d7081683078d Mon Sep 17 00:00:00 2001 From: Jacob Gelman <3182119+ladvoc@users.noreply.github.com> Date: Mon, 19 Oct 2026 11:01:35 -0700 Subject: [PATCH 15/23] Don't alias datatrack package --- datatracksender.go | 12 ++++++------ datatracksender_test.go | 6 +++--- engine.go | 4 ++-- localdatatrack.go | 4 ++-- localparticipant.go | 4 ++-- room.go | 6 +++--- 6 files changed, 18 insertions(+), 18 deletions(-) diff --git a/datatracksender.go b/datatracksender.go index 1c95dacc..7b1f464a 100644 --- a/datatracksender.go +++ b/datatracksender.go @@ -20,7 +20,7 @@ import ( protoLogger "github.com/livekit/protocol/logger" "github.com/pion/webrtc/v4" - dtbackend "github.com/livekit/server-sdk-go/v2/internal/datatrack" + "github.com/livekit/server-sdk-go/v2/internal/datatrack" ) const dataTrackBufferedAmountLowThreshold = 8 * 1024 @@ -32,7 +32,7 @@ type dataTrackSender struct { lock sync.Mutex dc *webrtc.DataChannel - frame dtbackend.FramePackets + frame datatrack.FramePackets notify chan struct{} done chan struct{} } @@ -60,7 +60,7 @@ func (s *dataTrackSender) setLogger(log protoLogger.Logger) { s.log = log } -func (s *dataTrackSender) send(frame dtbackend.FramePackets) { +func (s *dataTrackSender) send(frame datatrack.FramePackets) { if dropped := s.push(frame); dropped != nil { s.log.Debugw("dropping data track frame", "numPackets", len(dropped)) } @@ -77,7 +77,7 @@ func (s *dataTrackSender) stop() { close(s.done) } -func (s *dataTrackSender) push(frame dtbackend.FramePackets) (dropped dtbackend.FramePackets) { +func (s *dataTrackSender) push(frame datatrack.FramePackets) (dropped datatrack.FramePackets) { if len(frame) == 0 { return nil } @@ -90,7 +90,7 @@ func (s *dataTrackSender) push(frame dtbackend.FramePackets) (dropped dtbackend. return dropped } -func (s *dataTrackSender) pop() dtbackend.FramePackets { +func (s *dataTrackSender) pop() datatrack.FramePackets { s.lock.Lock() defer s.lock.Unlock() @@ -102,7 +102,7 @@ func (s *dataTrackSender) pop() dtbackend.FramePackets { func (s *dataTrackSender) run() { var ( dc *webrtc.DataChannel - inFlight dtbackend.FramePackets + inFlight datatrack.FramePackets ) for { select { diff --git a/datatracksender_test.go b/datatracksender_test.go index a572e86d..f2767b2b 100644 --- a/datatracksender_test.go +++ b/datatracksender_test.go @@ -19,11 +19,11 @@ import ( "github.com/stretchr/testify/require" - dtbackend "github.com/livekit/server-sdk-go/v2/internal/datatrack" + "github.com/livekit/server-sdk-go/v2/internal/datatrack" ) -func testFrame(marker byte, packets int) dtbackend.FramePackets { - frame := make(dtbackend.FramePackets, packets) +func testFrame(marker byte, packets int) datatrack.FramePackets { + frame := make(datatrack.FramePackets, packets) for i := range frame { frame[i] = []byte{marker, byte(i)} } diff --git a/engine.go b/engine.go index 1bd5415d..4ce6c9c4 100644 --- a/engine.go +++ b/engine.go @@ -34,7 +34,7 @@ import ( protosignalling "github.com/livekit/protocol/signalling" "github.com/livekit/server-sdk-go/v2/e2ee" - dtbackend "github.com/livekit/server-sdk-go/v2/internal/datatrack" + "github.com/livekit/server-sdk-go/v2/internal/datatrack" "github.com/livekit/server-sdk-go/v2/signalling" ) @@ -1866,6 +1866,6 @@ func waitUntilConnected(ctx context.Context, d time.Duration, test func() bool) } } -func (e *RTCEngine) sendDataTrackFrame(frame dtbackend.FramePackets) { +func (e *RTCEngine) sendDataTrackFrame(frame datatrack.FramePackets) { e.dataTrackSender.send(frame) } diff --git a/localdatatrack.go b/localdatatrack.go index 2717951a..c94ac405 100644 --- a/localdatatrack.go +++ b/localdatatrack.go @@ -21,7 +21,7 @@ import ( "github.com/livekit/protocol/livekit" "github.com/livekit/server-sdk-go/v2/datatrack" - dtbackend "github.com/livekit/server-sdk-go/v2/internal/datatrack" + dtinternal "github.com/livekit/server-sdk-go/v2/internal/datatrack" ) const dataTrackPublishTimeout = 10 * time.Second @@ -42,7 +42,7 @@ func (t localDataTrackTransport) SendUnpublishRequest(req *livekit.UnpublishData return t.engine.SendUnpublishDataTrack(req) } -func (t localDataTrackTransport) SendFrame(frame dtbackend.FramePackets) { +func (t localDataTrackTransport) SendFrame(frame dtinternal.FramePackets) { t.engine.sendDataTrackFrame(frame) } diff --git a/localparticipant.go b/localparticipant.go index 1cc8b6b3..7b0a80f6 100644 --- a/localparticipant.go +++ b/localparticipant.go @@ -30,7 +30,7 @@ import ( "github.com/livekit/protocol/livekit" protoLogger "github.com/livekit/protocol/logger" - dtbackend "github.com/livekit/server-sdk-go/v2/internal/datatrack" + "github.com/livekit/server-sdk-go/v2/internal/datatrack" ) const ( @@ -40,7 +40,7 @@ const ( type LocalParticipant struct { baseParticipant engine *RTCEngine - dataTracks *dtbackend.LocalManager + dataTracks *datatrack.LocalManager subscriptionPermission *livekit.SubscriptionPermission serverInfo *livekit.ServerInfo diff --git a/room.go b/room.go index 0fb18e51..252e6ab0 100644 --- a/room.go +++ b/room.go @@ -35,7 +35,7 @@ import ( protosignalling "github.com/livekit/protocol/signalling" "github.com/livekit/server-sdk-go/v2/e2ee" - dtbackend "github.com/livekit/server-sdk-go/v2/internal/datatrack" + "github.com/livekit/server-sdk-go/v2/internal/datatrack" "github.com/livekit/server-sdk-go/v2/signalling" "github.com/livekit/mediatransportutil/pkg/pacer" @@ -323,7 +323,7 @@ type Room struct { sid string name string LocalParticipant *LocalParticipant - localDataTracks *dtbackend.LocalManager + localDataTracks *datatrack.LocalManager callback *RoomCallback sidReady chan struct{} disconnectReason livekit.DisconnectReason @@ -368,7 +368,7 @@ func NewRoom(callback *RoomCallback) *Room { r.engine = NewRTCEngine(r.useSinglePeerConnection, r, r.getLocalParticipantSID, r.regionURLProvider) r.LocalParticipant = newLocalParticipant(r.engine, r.callback, r.serverInfo, r.log) - r.localDataTracks = dtbackend.NewLocalManager(dtbackend.LocalManagerParams{Transport: localDataTrackTransport{engine: r.engine}, Logger: r.log}) + r.localDataTracks = datatrack.NewLocalManager(datatrack.LocalManagerParams{Transport: localDataTrackTransport{engine: r.engine}, Logger: r.log}) r.LocalParticipant.dataTracks = r.localDataTracks return r } From fba89d622252bd6cd1453addc27b96c985b9eefe Mon Sep 17 00:00:00 2001 From: Jacob Gelman <3182119+ladvoc@users.noreply.github.com> Date: Mon, 19 Oct 2026 11:07:44 -0700 Subject: [PATCH 16/23] Expose set logger for local manager --- internal/datatrack/local.go | 28 ++++++++++++++++++++-------- localparticipant.go | 1 + room.go | 2 +- 3 files changed, 22 insertions(+), 9 deletions(-) diff --git a/internal/datatrack/local.go b/internal/datatrack/local.go index 11c28bc2..fde31e96 100644 --- a/internal/datatrack/local.go +++ b/internal/datatrack/local.go @@ -80,6 +80,7 @@ type LocalManagerParams struct { // concurrent use. type LocalManager struct { params LocalManagerParams + log atomic.Pointer[logger.Logger] mu sync.Mutex handles handleAllocator @@ -96,11 +97,22 @@ func NewLocalManager(params LocalManagerParams) *LocalManager { if params.Logger == nil { params.Logger = logger.GetLogger() } - return &LocalManager{ + m := &LocalManager{ params: params, pending: make(map[trackHandle]chan publishResult), active: make(map[trackHandle]*LocalTrack), } + m.SetLogger(params.Logger) + return m +} + +// SetLogger replaces the logger given at construction. +func (m *LocalManager) SetLogger(l logger.Logger) { + m.log.Store(&l) +} + +func (m *LocalManager) logger() logger.Logger { + return *m.log.Load() } // Publish requests a publication and waits for the SFU's answer. Ending ctx abandons the request; @@ -163,7 +175,7 @@ func (m *LocalManager) Publish(ctx context.Context, options PublishOptions) (*Lo func (m *LocalManager) HandlePublishResponse(msg *livekit.PublishDataTrackResponse) { info, err := infoFromProto(msg.GetInfo()) if err != nil { - m.params.Logger.Warnw("ignoring invalid publish data track response", err) + m.logger().Warnw("ignoring invalid publish data track response", err) return } m.resolvePublish(publishResponse{handle: info.pubHandle, info: info}) @@ -198,12 +210,12 @@ func (m *LocalManager) resolvePublish(response publishResponse) { switch { case isActive && response.err != nil: - m.params.Logger.Warnw("republish failed for data track", response.err, "handle", response.handle) + m.logger().Warnw("republish failed for data track", response.err, "handle", response.handle) case isActive: if track.completeRepublish(response.info.SID) { - m.params.Logger.Debugw("data track republished", "handle", response.handle, "sid", response.info.SID) + m.logger().Debugw("data track republished", "handle", response.handle, "sid", response.info.SID) } else { - m.params.Logger.Warnw("data track already published", nil, "handle", response.handle) + m.logger().Warnw("data track already published", nil, "handle", response.handle) } case response.err == nil: // accepted after the request was abandoned, unpublish to keep the SFU consistent @@ -215,7 +227,7 @@ func (m *LocalManager) resolvePublish(response publishResponse) { func (m *LocalManager) HandleUnpublishResponse(msg *livekit.UnpublishDataTrackResponse) { handle, err := handleFromUint32(msg.GetInfo().GetPubHandle()) if err != nil { - m.params.Logger.Warnw("ignoring invalid unpublish data track response", err) + m.logger().Warnw("ignoring invalid unpublish data track response", err) return } @@ -247,7 +259,7 @@ func (m *LocalManager) unpublish(handle trackHandle) { func (m *LocalManager) sendUnpublish(handle trackHandle) { if err := m.params.Transport.SendUnpublishRequest(unpublishRequestToProto(handle)); err != nil { - m.params.Logger.Debugw("could not send unpublish data track request", "handle", handle, "error", err) + m.logger().Debugw("could not send unpublish data track request", "handle", handle, "error", err) } } @@ -274,7 +286,7 @@ func (m *LocalManager) RepublishTracks() { for _, request := range requests { if err := m.params.Transport.SendPublishRequest(context.Background(), request.toProto()); err != nil { - m.params.Logger.Warnw("could not republish data track", err, "handle", request.handle) + m.logger().Warnw("could not republish data track", err, "handle", request.handle) } } } diff --git a/localparticipant.go b/localparticipant.go index 7b0a80f6..f507fd8f 100644 --- a/localparticipant.go +++ b/localparticipant.go @@ -1208,4 +1208,5 @@ func (p *LocalParticipant) SetLogger(log protoLogger.Logger) { pub.SetLogger(p.log) return true }) + p.dataTracks.SetLogger(p.log) } diff --git a/room.go b/room.go index 252e6ab0..fb13ac9a 100644 --- a/room.go +++ b/room.go @@ -368,7 +368,7 @@ func NewRoom(callback *RoomCallback) *Room { r.engine = NewRTCEngine(r.useSinglePeerConnection, r, r.getLocalParticipantSID, r.regionURLProvider) r.LocalParticipant = newLocalParticipant(r.engine, r.callback, r.serverInfo, r.log) - r.localDataTracks = datatrack.NewLocalManager(datatrack.LocalManagerParams{Transport: localDataTrackTransport{engine: r.engine}, Logger: r.log}) + r.localDataTracks = datatrack.NewLocalManager(datatrack.LocalManagerParams{Transport: localDataTrackTransport{engine: r.engine}, Logger: r.LocalParticipant.log}) r.LocalParticipant.dataTracks = r.localDataTracks return r } From 0bbd109cd5f1ae552ad105ef83bc2178a381de53 Mon Sep 17 00:00:00 2001 From: Jacob Gelman <3182119+ladvoc@users.noreply.github.com> Date: Mon, 19 Oct 2026 11:11:03 -0700 Subject: [PATCH 17/23] Relocate local manager to local participant --- localparticipant.go | 4 +++- room.go | 16 ++++++---------- 2 files changed, 9 insertions(+), 11 deletions(-) diff --git a/localparticipant.go b/localparticipant.go index f507fd8f..8f4a37e9 100644 --- a/localparticipant.go +++ b/localparticipant.go @@ -49,13 +49,15 @@ type LocalParticipant struct { } func newLocalParticipant(engine *RTCEngine, roomcallback *RoomCallback, serverInfo *livekit.ServerInfo, log protoLogger.Logger) *LocalParticipant { - return &LocalParticipant{ + p := &LocalParticipant{ baseParticipant: *newBaseParticipant(roomcallback, log.WithValues("isLocal", true)), engine: engine, serverInfo: serverInfo, rpcPendingAcks: &sync.Map{}, rpcPendingResponses: &sync.Map{}, } + p.dataTracks = datatrack.NewLocalManager(datatrack.LocalManagerParams{Transport: localDataTrackTransport{engine: engine}, Logger: p.log}) + return p } func (p *LocalParticipant) prepareTrackPublication(track webrtc.TrackLocal, opts *TrackPublicationOptions, pubOpts ...LocalTrackPublishOption) (*LocalTrackPublication, *livekit.AddTrackRequest, error) { diff --git a/room.go b/room.go index fb13ac9a..acbef166 100644 --- a/room.go +++ b/room.go @@ -35,7 +35,6 @@ import ( protosignalling "github.com/livekit/protocol/signalling" "github.com/livekit/server-sdk-go/v2/e2ee" - "github.com/livekit/server-sdk-go/v2/internal/datatrack" "github.com/livekit/server-sdk-go/v2/signalling" "github.com/livekit/mediatransportutil/pkg/pacer" @@ -323,7 +322,6 @@ type Room struct { sid string name string LocalParticipant *LocalParticipant - localDataTracks *datatrack.LocalManager callback *RoomCallback sidReady chan struct{} disconnectReason livekit.DisconnectReason @@ -368,8 +366,6 @@ func NewRoom(callback *RoomCallback) *Room { r.engine = NewRTCEngine(r.useSinglePeerConnection, r, r.getLocalParticipantSID, r.regionURLProvider) r.LocalParticipant = newLocalParticipant(r.engine, r.callback, r.serverInfo, r.log) - r.localDataTracks = datatrack.NewLocalManager(datatrack.LocalManagerParams{Transport: localDataTrackTransport{engine: r.engine}, Logger: r.LocalParticipant.log}) - r.LocalParticipant.dataTracks = r.localDataTracks return r } @@ -782,7 +778,7 @@ func (r *Room) sendSyncState() { Subscribe: !sendUnsub, }, PublishTracks: publishedTracks, - PublishDataTracks: r.localDataTracks.PublishResponsesForSyncState(), + PublishDataTracks: r.LocalParticipant.dataTracks.PublishResponsesForSyncState(), DataChannels: dataChannels, TrackSidsDisabled: trackSidsDisabled, // MIGRATION-TODO DatachannelReceiveStates @@ -792,7 +788,7 @@ func (r *Room) sendSyncState() { func (r *Room) cleanup() { r.engine.Close() r.LocalParticipant.closeTracks() - r.localDataTracks.Shutdown() + r.LocalParticipant.dataTracks.Shutdown() r.setSid("", true) r.byteStreamHandlers.Clear() @@ -1112,7 +1108,7 @@ func (r *Room) OnRestarted( r.OnParticipantUpdate(otherParticipants) r.LocalParticipant.republishTracks() - r.localDataTracks.RepublishTracks() + r.LocalParticipant.dataTracks.RepublishTracks() } func (r *Room) OnRestartConnected() { @@ -1408,7 +1404,7 @@ func (r *Room) OnMediaSectionsRequirement(mediaSectionsRequirement *livekit.Medi // OnRequestResponse handles responses to signal requests that no caller is waiting on. func (r *Room) OnRequestResponse(res *livekit.RequestResponse) { - if r.localDataTracks.HandleRequestResponse(res) { + if r.LocalParticipant.dataTracks.HandleRequestResponse(res) { return } if res.GetReason() != livekit.RequestResponse_OK { @@ -1422,11 +1418,11 @@ func (r *Room) OnRequestResponse(res *livekit.RequestResponse) { } func (r *Room) OnPublishDataTrackResponse(publishDataTrackResponse *livekit.PublishDataTrackResponse) { - r.localDataTracks.HandlePublishResponse(publishDataTrackResponse) + r.LocalParticipant.dataTracks.HandlePublishResponse(publishDataTrackResponse) } func (r *Room) OnUnpublishDataTrackResponse(unpublishDataTrackResponse *livekit.UnpublishDataTrackResponse) { - r.localDataTracks.HandleUnpublishResponse(unpublishDataTrackResponse) + r.LocalParticipant.dataTracks.HandleUnpublishResponse(unpublishDataTrackResponse) } func (r *Room) OnDataTrackSubscriberHandles(dataTrackSubscriberHandles *livekit.DataTrackSubscriberHandles) { From 6d411e9da57ad8cf17515bf92b280fa77f79519e Mon Sep 17 00:00:00 2001 From: Jacob Gelman <3182119+ladvoc@users.noreply.github.com> Date: Mon, 19 Oct 2026 11:14:59 -0700 Subject: [PATCH 18/23] Lower bound on test timeout --- localdatatrack_test.go | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/localdatatrack_test.go b/localdatatrack_test.go index 60e4ebb3..95bcdacd 100644 --- a/localdatatrack_test.go +++ b/localdatatrack_test.go @@ -28,11 +28,12 @@ func TestPublishDataTrackDeadlineInterruptsConnectionWait(t *testing.T) { room := NewRoom(&RoomCallback{}) t.Cleanup(room.engine.Close) - ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + const deadline = 50 * time.Millisecond + start := time.Now() + ctx, cancel := context.WithTimeout(context.Background(), deadline) defer cancel() - start := time.Now() _, err := room.LocalParticipant.PublishDataTrack(ctx, "test") require.ErrorIs(t, err, context.DeadlineExceeded) - require.Less(t, time.Since(start), time.Second) + require.WithinRange(t, time.Now(), start.Add(deadline), start.Add(time.Second)) } From c008b8332b8f0ee4a8f73ce4fc7c05d4d3f5ab7a Mon Sep 17 00:00:00 2001 From: Jacob Gelman <3182119+ladvoc@users.noreply.github.com> Date: Mon, 19 Oct 2026 11:17:09 -0700 Subject: [PATCH 19/23] Explicit unlock at end of test --- internal/datatrack/local_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/internal/datatrack/local_test.go b/internal/datatrack/local_test.go index a7fffdf4..b813948c 100644 --- a/internal/datatrack/local_test.go +++ b/internal/datatrack/local_test.go @@ -233,9 +233,9 @@ func TestLocalManager_PublishCancelledWhileTransportBlocked(t *testing.T) { // The abandoned request released its handle and left nothing pending m.mu.Lock() - defer m.mu.Unlock() require.Empty(t, m.pending) require.Empty(t, m.active) + m.mu.Unlock() } func TestLocalManager_PublishWithE2EE(t *testing.T) { From 89d603a602c0a8fab298034a6e63a7e22a70a99d Mon Sep 17 00:00:00 2001 From: Jacob Gelman <3182119+ladvoc@users.noreply.github.com> Date: Mon, 19 Oct 2026 11:59:03 -0700 Subject: [PATCH 20/23] Mark data track unpublished after republish fail --- internal/datatrack/local.go | 7 ++++++- internal/datatrack/local_test.go | 33 ++++++++++++++++++++++++++++++++ 2 files changed, 39 insertions(+), 1 deletion(-) diff --git a/internal/datatrack/local.go b/internal/datatrack/local.go index fde31e96..f4ad3d37 100644 --- a/internal/datatrack/local.go +++ b/internal/datatrack/local.go @@ -206,10 +206,14 @@ func (m *LocalManager) resolvePublish(response publishResponse) { return } track, isActive := m.active[response.handle] + if isActive && response.err != nil { + delete(m.active, response.handle) + } m.mu.Unlock() switch { case isActive && response.err != nil: + track.markUnpublished() m.logger().Warnw("republish failed for data track", response.err, "handle", response.handle) case isActive: if track.completeRepublish(response.info.SID) { @@ -264,7 +268,8 @@ func (m *LocalManager) sendUnpublish(handle trackHandle) { } // RepublishTracks re-requests every publication after a full reconnect. Pending publications fail -// with ErrDisconnected; published tracks drop frames until the SFU answers with their new SID. +// with ErrDisconnected; published tracks drop frames until the SFU answers with their new SID, and +// are unpublished if it rejects them. func (m *LocalManager) RepublishTracks() { m.mu.Lock() for handle, result := range m.pending { diff --git a/internal/datatrack/local_test.go b/internal/datatrack/local_test.go index b813948c..b98660b9 100644 --- a/internal/datatrack/local_test.go +++ b/internal/datatrack/local_test.go @@ -304,6 +304,39 @@ func TestLocalManager_RepublishTracks(t *testing.T) { require.NoError(t, track.TryPush(Frame{Payload: []byte{0xff}})) } +func TestLocalManager_RepublishRejected(t *testing.T) { + transport := newFakeLocalTransport() + m := NewLocalManager(LocalManagerParams{Transport: transport}) + + result := publishAsync(context.Background(), m, "track") + request := expectEvent(t, transport.publishRequests) + m.HandlePublishResponse(accepted(request, "DTR_1234")) + track := expectEvent(t, result).track + + // Simulate reconnect + m.RepublishTracks() + request = expectEvent(t, transport.publishRequests) + + // SFU rejects the republish + handled := m.HandleRequestResponse(&livekit.RequestResponse{ + Request: &livekit.RequestResponse_PublishDataTrack{PublishDataTrack: request}, + Reason: livekit.RequestResponse_LIMIT_EXCEEDED, + }) + require.True(t, handled) + + // Track is unpublished and no longer republished on the next reconnect + expectEvent(t, track.Unpublished()) + require.False(t, track.IsPublished()) + require.ErrorIs(t, track.TryPush(Frame{Payload: []byte{0xff}}), ErrUnpublished) + + m.RepublishTracks() + select { + case request := <-transport.publishRequests: + require.Failf(t, "unexpected republish", "handle %d", request.GetPubHandle()) + case <-time.After(50 * time.Millisecond): + } +} + func TestLocalManager_QueryPublished(t *testing.T) { transport := newFakeLocalTransport() m := NewLocalManager(LocalManagerParams{Transport: transport}) From 953bf0c1d5028f125c781e0c4078c9b1db78eddd Mon Sep 17 00:00:00 2001 From: Jacob Gelman <3182119+ladvoc@users.noreply.github.com> Date: Mon, 19 Oct 2026 12:04:51 -0700 Subject: [PATCH 21/23] Use single mutex for sync inside local manager Remove seperate atomic --- internal/datatrack/local.go | 44 +++++++++++++++++++++++-------------- 1 file changed, 27 insertions(+), 17 deletions(-) diff --git a/internal/datatrack/local.go b/internal/datatrack/local.go index f4ad3d37..fda134c5 100644 --- a/internal/datatrack/local.go +++ b/internal/datatrack/local.go @@ -327,7 +327,7 @@ func (m *LocalManager) Shutdown() { } } -type publishState int32 +type publishState int const ( statePublished publishState = iota @@ -338,13 +338,12 @@ const ( // LocalTrack is a data track published by the local participant. Methods are safe for concurrent // use. type LocalTrack struct { - manager *LocalManager - handle trackHandle - state atomic.Int32 - unpublished chan struct{} - unpublishOnce sync.Once + manager *LocalManager + handle trackHandle + unpublished chan struct{} mu sync.Mutex + state publishState info Info pipeline localPipeline } @@ -374,7 +373,9 @@ func (t *LocalTrack) Info() Info { // IsPublished reports whether the track is published. A track being republished after a // reconnect still counts as published. func (t *LocalTrack) IsPublished() bool { - return publishState(t.state.Load()) != stateUnpublished + t.mu.Lock() + defer t.mu.Unlock() + return t.state != stateUnpublished } // Unpublished is closed once the track is no longer published, whether by Unpublish, the SFU, or @@ -387,14 +388,15 @@ func (t *LocalTrack) Unpublished() <-chan struct{} { // unpublished and ErrQueueFull while it is being republished after a reconnect. Frames are never // retained, so the caller may reuse the payload immediately. func (t *LocalTrack) TryPush(frame Frame) error { - switch publishState(t.state.Load()) { + t.mu.Lock() + switch t.state { case stateRepublishing: + t.mu.Unlock() return ErrQueueFull case stateUnpublished: + t.mu.Unlock() return ErrUnpublished } - - t.mu.Lock() packets, err := t.pipeline.processFrame(frame) t.mu.Unlock() if err != nil { @@ -410,23 +412,31 @@ func (t *LocalTrack) Unpublish() { } func (t *LocalTrack) markUnpublished() { - t.state.Store(int32(stateUnpublished)) - t.unpublishOnce.Do(func() { close(t.unpublished) }) + t.mu.Lock() + defer t.mu.Unlock() + if t.state == stateUnpublished { + return + } + t.state = stateUnpublished + close(t.unpublished) } func (t *LocalTrack) beginRepublish() Info { - t.state.Store(int32(stateRepublishing)) - return t.Info() + t.mu.Lock() + defer t.mu.Unlock() + t.state = stateRepublishing + return t.info } // completeRepublish reports false if the track was not being republished. func (t *LocalTrack) completeRepublish(sid livekit.TrackID) bool { - if !t.state.CompareAndSwap(int32(stateRepublishing), int32(statePublished)) { + t.mu.Lock() + defer t.mu.Unlock() + if t.state != stateRepublishing { return false } - t.mu.Lock() + t.state = statePublished t.info.SID = sid - t.mu.Unlock() return true } From d905ce772d7b13c742d2212771943e88177dbb48 Mon Sep 17 00:00:00 2001 From: Jacob Gelman <3182119+ladvoc@users.noreply.github.com> Date: Tue, 20 Oct 2026 11:51:19 -0700 Subject: [PATCH 22/23] Don't send requests after cancellation --- internal/datatrack/local.go | 3 +++ internal/datatrack/local_test.go | 11 +++++++++++ 2 files changed, 14 insertions(+) diff --git a/internal/datatrack/local.go b/internal/datatrack/local.go index fda134c5..44e4c138 100644 --- a/internal/datatrack/local.go +++ b/internal/datatrack/local.go @@ -128,6 +128,9 @@ func (m *LocalManager) Publish(ctx context.Context, options PublishOptions) (*Lo if err := validateSchema(options.FrameEncoding, schemaEncoding); err != nil { return nil, err } + if err := ctx.Err(); err != nil { + return nil, err + } m.mu.Lock() handle, err := m.handles.get() diff --git a/internal/datatrack/local_test.go b/internal/datatrack/local_test.go index b98660b9..e0b76941 100644 --- a/internal/datatrack/local_test.go +++ b/internal/datatrack/local_test.go @@ -213,6 +213,17 @@ func TestLocalManager_PublishCancelled(t *testing.T) { require.Equal(t, request.GetPubHandle(), unpublish.GetPubHandle()) } +func TestLocalManager_PublishAlreadyCancelled(t *testing.T) { + transport := newFakeLocalTransport() + m := NewLocalManager(LocalManagerParams{Transport: transport}) + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + _, err := m.Publish(ctx, PublishOptions{Name: "test"}) + require.ErrorIs(t, err, context.Canceled) + require.Empty(t, transport.publishRequests) +} + // blockingTransport never becomes ready: SendPublishRequest waits on ctx like the real transport // waits on the engine connection. type blockingTransport struct{ *fakeLocalTransport } From 81fb6a13263fa8f8090fc03f0d58cdbb82408ea1 Mon Sep 17 00:00:00 2001 From: Jacob Gelman <3182119+ladvoc@users.noreply.github.com> Date: Tue, 20 Oct 2026 11:52:08 -0700 Subject: [PATCH 23/23] Note empty payloads are dropped. --- internal/datatrack/local.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/internal/datatrack/local.go b/internal/datatrack/local.go index 44e4c138..3437c151 100644 --- a/internal/datatrack/local.go +++ b/internal/datatrack/local.go @@ -389,7 +389,7 @@ func (t *LocalTrack) Unpublished() <-chan struct{} { // TryPush sends a frame to the track's subscribers. It returns ErrUnpublished once the track is // unpublished and ErrQueueFull while it is being republished after a reconnect. Frames are never -// retained, so the caller may reuse the payload immediately. +// retained, so the caller may reuse the payload immediately. Frames with empty payloads are dropped. func (t *LocalTrack) TryPush(frame Frame) error { t.mu.Lock() switch t.state {