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 <noreply@anthropic.com>

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

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

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

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 {

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