From 6a64df20df7a0cef6cbe17d305347387659f180c Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Thu, 20 Aug 2026 13:51:10 +0530 Subject: [PATCH] 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 | 53 +++++++++++++++++++++++++++- pkg/rtc/participant_internal_test.go | 48 +++++++++++++++++++++++++ 2 files changed, 100 insertions(+), 1 deletion(-) diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 8aab59858..daa68dde4 100644 --- a/pkg/rtc/participant.go +++ b/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() diff --git a/pkg/rtc/participant_internal_test.go b/pkg/rtc/participant_internal_test.go index 33263a360..2cde9181c 100644 --- a/pkg/rtc/participant_internal_test.go +++ b/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) + }) +}