Add a small cache for data messages received via SendData API. (#4781)

SendData API messages do not have sender ID or sequence number. So, they
were not cached and hence excluded from the reliable caching feature
which is meant to provide reliability of data channel messages between
the time other participants see participant as ACTIVE (which happens on
ICE connected) and data channel being open (DTLS done + data channels
opened).

Add a small cache for that and flush those messages on data channel
establishment.

pkg/rtc/participant.go

@@ -81,6 +81,10 @@ const (
 	cMaxPendingTracks       = 20
 	cMaxPendingQueuedTracks = 3
 
+	// unsequenced reliable data (server API sends) cannot be recovered from the
+	// data message cache, so it is held here until the reliable data channel is writable
+	cMaxJoiningUnsequencedReliableBytes = 100_000
+
 	PingIntervalSeconds = 5
 	PingTimeoutSeconds  = 15
 )
@@ -150,6 +154,9 @@ type reliableDataInfo struct {
 	joiningMessageLock            sync.Mutex
 	joiningMessageFirstSeqs       map[livekit.ParticipantID]uint32
 	joiningMessageLastWrittenSeqs map[livekit.ParticipantID]uint32
+	joiningUnsequencedMessages    [][]byte
+	joiningUnsequencedBytes       int
+	joiningUnsequencedDropped     int
 	lastPubReliableSeq            atomic.Uint32
 	stopReliableByMigrateOut      atomic.Bool
 	canWriteReliable              bool
@@ -3956,7 +3963,37 @@ func (p *ParticipantImpl) SupportsTransceiverReuse(mt types.MediaTrack) bool {
 }
 
 func (p *ParticipantImpl) SendDataMessage(kind livekit.DataPacket_Kind, data []byte, sender livekit.ParticipantID, seq uint32) error {
-	if sender == "" || kind != livekit.DataPacket_RELIABLE || seq == 0 {
+	if kind != livekit.DataPacket_RELIABLE {
+		if p.State() != livekit.ParticipantInfo_ACTIVE {
+			return ErrDataChannelUnavailable
+		}
+		return p.TransportManager.SendDataMessage(kind, data)
+	}
+
+	if sender == "" || seq == 0 {
+		// Unsequenced reliable data, i. e. not published by a participant, room service
+		// SendData for example. Such a message cannot be recovered by
+		// replayJoiningReliableMessages as the data message cache is keyed on
+		// sender/sequence number, so hold on to the message itself here till the
+		// reliable data channel is writable.
+		p.reliableDataInfo.joiningMessageLock.Lock()
+		if !p.reliableDataInfo.canWriteReliable {
+			if p.reliableDataInfo.joiningUnsequencedBytes+len(data) > cMaxJoiningUnsequencedReliableBytes {
+				p.reliableDataInfo.joiningUnsequencedDropped++
+				p.reliableDataInfo.joiningMessageLock.Unlock()
+				return ErrDataChannelUnavailable
+			}
+
+			p.reliableDataInfo.joiningUnsequencedMessages = append(
+				p.reliableDataInfo.joiningUnsequencedMessages,
+				slices.Clone(data),
+			)
+			p.reliableDataInfo.joiningUnsequencedBytes += len(data)
+			p.reliableDataInfo.joiningMessageLock.Unlock()
+			return nil
+		}
+		p.reliableDataInfo.joiningMessageLock.Unlock()
+
 		if p.State() != livekit.ParticipantInfo_ACTIVE {
 			return ErrDataChannelUnavailable
 		}
@@ -4071,6 +4108,20 @@ func (p *ParticipantImpl) replayJoiningReliableMessages() {
 		p.TransportManager.SendDataMessage(livekit.DataPacket_RELIABLE, msgCache.Data)
 	}
 
+	for _, msg := range p.reliableDataInfo.joiningUnsequencedMessages {
+		p.TransportManager.SendDataMessage(livekit.DataPacket_RELIABLE, msg)
+	}
+	if p.reliableDataInfo.joiningUnsequencedDropped != 0 {
+		p.params.Logger.Warnw(
+			"dropped unsequenced reliable data messages while joining", nil,
+			"numDropped", p.reliableDataInfo.joiningUnsequencedDropped,
+			"numReplayed", len(p.reliableDataInfo.joiningUnsequencedMessages),
+		)
+	}
+	p.reliableDataInfo.joiningUnsequencedMessages = nil
+	p.reliableDataInfo.joiningUnsequencedBytes = 0
+	p.reliableDataInfo.joiningUnsequencedDropped = 0
+
 	p.reliableDataInfo.joiningMessageFirstSeqs = make(map[livekit.ParticipantID]uint32)
 	p.reliableDataInfo.canWriteReliable = true
 	p.reliableDataInfo.joiningMessageLock.Unlock()

pkg/rtc/participant_internal_test.go

@@ -847,3 +847,51 @@ func newParticipantForTestWithOpts(identity livekit.ParticipantIdentity, opts *p
 func newParticipantForTest(identity livekit.ParticipantIdentity) *ParticipantImpl {
 	return newParticipantForTestWithOpts(identity, nil)
 }
+
+func TestUnsequencedReliableDataBufferedWhileJoining(t *testing.T) {
+	// unsequenced reliable data, i. e. room service SendData, arriving before the
+	// reliable data channel is writable should be held and replayed, not dropped
+	t.Run("buffers and replays", func(t *testing.T) {
+		p := newParticipantForTest("test")
+		require.False(t, p.reliableDataInfo.canWriteReliable)
+
+		require.NoError(t, p.SendDataMessage(livekit.DataPacket_RELIABLE, []byte("one"), "", 0))
+		require.NoError(t, p.SendDataMessage(livekit.DataPacket_RELIABLE, []byte("two"), "", 0))
+
+		require.Equal(t, [][]byte{[]byte("one"), []byte("two")}, p.reliableDataInfo.joiningUnsequencedMessages)
+		require.Equal(t, 6, p.reliableDataInfo.joiningUnsequencedBytes)
+
+		p.replayJoiningReliableMessages()
+
+		require.True(t, p.reliableDataInfo.canWriteReliable)
+		require.Empty(t, p.reliableDataInfo.joiningUnsequencedMessages)
+		require.Zero(t, p.reliableDataInfo.joiningUnsequencedBytes)
+	})
+
+	t.Run("does not buffer once writable", func(t *testing.T) {
+		p := newParticipantForTest("test")
+		p.replayJoiningReliableMessages()
+
+		// no data channel in test, so the write through fails rather than getting buffered
+		require.Error(t, p.SendDataMessage(livekit.DataPacket_RELIABLE, []byte("one"), "", 0))
+		require.Empty(t, p.reliableDataInfo.joiningUnsequencedMessages)
+	})
+
+	t.Run("bounded buffer", func(t *testing.T) {
+		p := newParticipantForTest("test")
+
+		data := make([]byte, cMaxJoiningUnsequencedReliableBytes)
+		require.NoError(t, p.SendDataMessage(livekit.DataPacket_RELIABLE, data, "", 0))
+		require.Error(t, p.SendDataMessage(livekit.DataPacket_RELIABLE, []byte("overflow"), "", 0))
+
+		require.Len(t, p.reliableDataInfo.joiningUnsequencedMessages, 1)
+		require.Equal(t, 1, p.reliableDataInfo.joiningUnsequencedDropped)
+	})
+
+	t.Run("lossy is not buffered", func(t *testing.T) {
+		p := newParticipantForTest("test")
+
+		require.Error(t, p.SendDataMessage(livekit.DataPacket_LOSSY, []byte("one"), "", 0))
+		require.Empty(t, p.reliableDataInfo.joiningUnsequencedMessages)
+	})
+}