From adb7cb4f29aedecd13a4175726c17f8c3a2264d7 Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Sat, 31 Oct 2026 16:16:57 +0530 Subject: [PATCH] rtc: hold unlabeled data for a participant whose reliable channel is opening (#5113) An unlabeled data message sent to a participant without an unlabeled channel is converted to a UserPacket and written to the reliable channel. That write failed while the channel was still opening, and the message was lost. The participant now builds the UserPacket and sends it through SendDataMessage, which holds unsequenced reliable data until the channel is writable. The transport keeps only the raw path. Co-authored-by: Claude Fable 5.1 --- pkg/rtc/participant.go | 19 ++++++++++++++--- pkg/rtc/participant_internal_test.go | 12 +++++++++++ pkg/rtc/transport.go | 32 ++++------------------------ pkg/rtc/transportmanager.go | 8 ++----- test/client/client.go | 2 +- 5 files changed, 35 insertions(+), 38 deletions(-) diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 690db92be..aaddcd81d 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -4278,11 +4278,24 @@ func (p *ParticipantImpl) SendDataMessage(kind livekit.DataPacket_Kind, data []b } func (p *ParticipantImpl) SendDataMessageUnlabeled(data []byte, useRaw bool, sender livekit.ParticipantIdentity) error { - if p.State() != livekit.ParticipantInfo_ACTIVE { - return ErrDataChannelUnavailable + if useRaw || p.params.UseOneShotSignallingMode { + if p.State() != livekit.ParticipantInfo_ACTIVE { + return ErrDataChannelUnavailable + } + return p.TransportManager.SendDataMessageUnlabeled(data) } - return p.TransportManager.SendDataMessageUnlabeled(data, useRaw, sender) + // a UserPacket on the reliable channel, held like any reliable message while that channel opens + dpData, err := proto.Marshal(&livekit.DataPacket{ + ParticipantIdentity: string(sender), + Value: &livekit.DataPacket_User{ + User: &livekit.UserPacket{Payload: data}, + }, + }) + if err != nil { + return err + } + return p.SendDataMessage(livekit.DataPacket_RELIABLE, dpData, "", 0) } func (p *ParticipantImpl) onDataSendError(err error) { diff --git a/pkg/rtc/participant_internal_test.go b/pkg/rtc/participant_internal_test.go index 4dfe5148c..04f296033 100644 --- a/pkg/rtc/participant_internal_test.go +++ b/pkg/rtc/participant_internal_test.go @@ -1325,6 +1325,18 @@ func TestUnsequencedReliableDataBufferedWhileJoining(t *testing.T) { require.Error(t, p.SendDataMessage(livekit.DataPacket_LOSSY, []byte("one"), "", 0)) require.Empty(t, p.reliableDataInfo.joiningUnsequencedMessages) }) + + t.Run("unlabeled data is buffered as a user packet", func(t *testing.T) { + p := newParticipantForTest("test") + + require.NoError(t, p.SendDataMessageUnlabeled([]byte("one"), false, "alice")) + + require.Len(t, p.reliableDataInfo.joiningUnsequencedMessages, 1) + dp := &livekit.DataPacket{} + require.NoError(t, proto.Unmarshal(p.reliableDataInfo.joiningUnsequencedMessages[0], dp)) + require.Equal(t, "alice", dp.ParticipantIdentity) + require.Equal(t, []byte("one"), dp.GetUser().GetPayload()) + }) } func TestMigratingInParticipantWaitsForReconnectResponse(t *testing.T) { diff --git a/pkg/rtc/transport.go b/pkg/rtc/transport.go index 1632ea508..d72a04ecb 100644 --- a/pkg/rtc/transport.go +++ b/pkg/rtc/transport.go @@ -1537,39 +1537,15 @@ func (t *PCTransport) SendDataMessage(kind livekit.DataPacket_Kind, data []byte) return t.sendDataMessage(dc, data) } -func (t *PCTransport) SendDataMessageUnlabeled(data []byte, useRaw bool, sender livekit.ParticipantIdentity) error { - convertToUserPacket := false +func (t *PCTransport) SendDataMessageUnlabeled(data []byte) error { var dc *datachannel.DataChannelWriter[*webrtc.DataChannel] t.lock.RLock() - if t.params.UseOneShotSignallingMode || useRaw { - if len(t.unlabeledDataChannels) > 0 { - // use the first unlabeled to send - dc = t.unlabeledDataChannels[0] - } - } else { - if t.reliableDC != nil { - dc = t.reliableDC - } else if t.lossyDC != nil { - dc = t.lossyDC - } - - convertToUserPacket = true + if len(t.unlabeledDataChannels) > 0 { + // use the first unlabeled to send + dc = t.unlabeledDataChannels[0] } t.lock.RUnlock() - if convertToUserPacket { - dpData, err := proto.Marshal(&livekit.DataPacket{ - ParticipantIdentity: string(sender), - Value: &livekit.DataPacket_User{ - User: &livekit.UserPacket{Payload: data}, - }, - }) - if err != nil { - return err - } - return t.sendDataMessage(dc, dpData) - } - return t.sendDataMessage(dc, data) } diff --git a/pkg/rtc/transportmanager.go b/pkg/rtc/transportmanager.go index fad77024f..2aac96fe2 100644 --- a/pkg/rtc/transportmanager.go +++ b/pkg/rtc/transportmanager.go @@ -376,13 +376,9 @@ func (t *TransportManager) SendDataMessage(kind livekit.DataPacket_Kind, data [] return t.handleSendDataResult(t.getTransport(true).SendDataMessage(kind, data), kind.String(), len(data)) } -func (t *TransportManager) SendDataMessageUnlabeled(data []byte, useRaw bool, sender livekit.ParticipantIdentity) error { +func (t *TransportManager) SendDataMessageUnlabeled(data []byte) error { // downstream data is sent via primary peer connection - return t.handleSendDataResult( - t.getTransport(true).SendDataMessageUnlabeled(data, useRaw, sender), - "unlabeled", - len(data), - ) + return t.handleSendDataResult(t.getTransport(true).SendDataMessageUnlabeled(data), "unlabeled", len(data)) } func (t *TransportManager) handleSendDataResult(err error, kind string, size int) error { diff --git a/test/client/client.go b/test/client/client.go index 66030dc4a..9f1e4a65c 100644 --- a/test/client/client.go +++ b/test/client/client.go @@ -1066,7 +1066,7 @@ func (c *RTCClient) PublishDataUnlabeled(data []byte) error { return err } - return c.publisher.SendDataMessageUnlabeled(data, true, "test") + return c.publisher.SendDataMessageUnlabeled(data) } func (c *RTCClient) PublishDataTrack() (writer TrackWriter, err error) {