diff --git a/go.mod b/go.mod index 31a9debe5..ab2bb0b79 100644 --- a/go.mod +++ b/go.mod @@ -23,7 +23,7 @@ require ( github.com/jxskiss/base62 v1.1.0 github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731 github.com/livekit/mediatransportutil v0.0.0-20250519131108-fb90f5acfded - github.com/livekit/protocol v1.39.1-0.20250609143527-8af5761cacc6 + github.com/livekit/protocol v1.39.1-0.20250610020455-4ba773363c7f github.com/livekit/psrpc v0.6.1-0.20250511053145-465289d72c3c github.com/mackerelio/go-osstat v0.2.5 github.com/magefile/mage v1.15.0 diff --git a/go.sum b/go.sum index 9a0218ee5..5e5c14284 100644 --- a/go.sum +++ b/go.sum @@ -169,8 +169,8 @@ github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731 h1:9x+U2HGLrSw5AT github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731/go.mod h1:Rs3MhFwutWhGwmY1VQsygw28z5bWcnEYmS1OG9OxjOQ= github.com/livekit/mediatransportutil v0.0.0-20250519131108-fb90f5acfded h1:ylZPdnlX1RW9Z15SD4mp87vT2D2shsk0hpLJwSPcq3g= github.com/livekit/mediatransportutil v0.0.0-20250519131108-fb90f5acfded/go.mod h1:mSNtYzSf6iY9xM3UX42VEI+STHvMgHmrYzEHPcdhB8A= -github.com/livekit/protocol v1.39.1-0.20250609143527-8af5761cacc6 h1:aibQ28Ewev0otGVQEj/WWcovh+lVgwLu1DDTs1F54oU= -github.com/livekit/protocol v1.39.1-0.20250609143527-8af5761cacc6/go.mod h1:6HPISM0bkTXTk9RIaQTCe0IDbomBPz7Jwp+N3w5sqL0= +github.com/livekit/protocol v1.39.1-0.20250610020455-4ba773363c7f h1:/CvQV9cfA6oIro/rEfX8YQSCu82s1PPaAwm94b7z7XE= +github.com/livekit/protocol v1.39.1-0.20250610020455-4ba773363c7f/go.mod h1:6HPISM0bkTXTk9RIaQTCe0IDbomBPz7Jwp+N3w5sqL0= github.com/livekit/psrpc v0.6.1-0.20250511053145-465289d72c3c h1:WwEr0YBejYbKzk8LSaO9h8h0G9MnE7shyDu8yXQWmEc= github.com/livekit/psrpc v0.6.1-0.20250511053145-465289d72c3c/go.mod h1:kmD+AZPkWu0MaXIMv57jhNlbiSZZ/Jx4bzlxBDVmJes= github.com/mackerelio/go-osstat v0.2.5 h1:+MqTbZUhoIt4m8qzkVoXUJg1EuifwlAJSk4Yl2GXh+o= diff --git a/pkg/rtc/migrationdatacache.go b/pkg/rtc/migrationdatacache.go new file mode 100644 index 000000000..8c44eac61 --- /dev/null +++ b/pkg/rtc/migrationdatacache.go @@ -0,0 +1,59 @@ +package rtc + +import ( + "time" + + "github.com/livekit/protocol/livekit" +) + +type MigrationDataCacheState int + +const ( + MigrationDataCacheStateWaiting MigrationDataCacheState = iota + MigrationDataCacheStateTimeout + MigrationDataCacheStateDone +) + +type MigrationDataCache struct { + lastSeq uint32 + pkts []*livekit.DataPacket + state MigrationDataCacheState + expiredAt time.Time +} + +func NewMigrationDataCache(lastSeq uint32, expiredAt time.Time) *MigrationDataCache { + return &MigrationDataCache{ + lastSeq: lastSeq, + expiredAt: expiredAt, + } +} + +// Add adds a message to the cache if there is a gap between the last sequence number and cached messages then return the cache State: +// - MigrationDataCacheStateWaiting: waiting for the next packet (lastSeq + 1) of last sequence from old node +// - MigrationDataCacheStateTimeout: the next packet is not received before the expiredAt, participant will +// continue to process the reliable messages, subscribers will see the gap after the publisher migration +// - MigrationDataCacheStateDone: the next packet is received, participant can continue to process the reliable messages +func (c *MigrationDataCache) Add(pkt *livekit.DataPacket) MigrationDataCacheState { + if c.state == MigrationDataCacheStateDone || c.state == MigrationDataCacheStateTimeout { + return c.state + } + + if pkt.Sequence <= c.lastSeq { + return c.state + } + + if pkt.Sequence == c.lastSeq+1 { + c.state = MigrationDataCacheStateDone + return c.state + } + + c.pkts = append(c.pkts, pkt) + if time.Now().After(c.expiredAt) { + c.state = MigrationDataCacheStateTimeout + } + return c.state +} + +func (c *MigrationDataCache) Get() []*livekit.DataPacket { + return c.pkts +} diff --git a/pkg/rtc/migrationdatacache_test.go b/pkg/rtc/migrationdatacache_test.go new file mode 100644 index 000000000..046b2ddfc --- /dev/null +++ b/pkg/rtc/migrationdatacache_test.go @@ -0,0 +1,38 @@ +package rtc + +import ( + "testing" + "time" + + "github.com/stretchr/testify/require" + + "github.com/livekit/protocol/livekit" +) + +func TestMigrationDataCache_Add(t *testing.T) { + expiredAt := time.Now().Add(100 * time.Millisecond) + cache := NewMigrationDataCache(10, expiredAt) + + pkt1 := &livekit.DataPacket{Sequence: 9} + state := cache.Add(pkt1) + require.Equal(t, MigrationDataCacheStateWaiting, state) + require.Empty(t, cache.Get()) + + pkt2 := &livekit.DataPacket{Sequence: 11} + state = cache.Add(pkt2) + require.Equal(t, MigrationDataCacheStateDone, state) + require.Empty(t, cache.Get()) + + pkt3 := &livekit.DataPacket{Sequence: 12} + state = cache.Add(pkt3) + require.Equal(t, MigrationDataCacheStateDone, state) + require.Empty(t, cache.Get()) + + cache2 := NewMigrationDataCache(20, time.Now().Add(10*time.Millisecond)) + pkt4 := &livekit.DataPacket{Sequence: 22} + time.Sleep(20 * time.Millisecond) + state = cache2.Add(pkt4) + require.Equal(t, MigrationDataCacheStateTimeout, state) + require.Len(t, cache2.Get(), 1) + require.Equal(t, uint32(22), cache2.Get()[0].Sequence) +} diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 38a17cf2c..6764e22f5 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -64,8 +64,9 @@ const ( sdBatchSize = 30 rttUpdateInterval = 5 * time.Second - disconnectCleanupDuration = 5 * time.Second - migrationWaitDuration = 3 * time.Second + disconnectCleanupDuration = 5 * time.Second + migrationWaitDuration = 3 * time.Second + migrationWaitContinuousMsgDuration = 2 * time.Second PingIntervalSeconds = 5 PingTimeoutSeconds = 15 @@ -110,6 +111,16 @@ func (p participantUpdateInfo) String() string { return fmt.Sprintf("identity: %s, version: %d, state: %s, updatedAt: %s", p.identity, p.version, p.state.String(), p.updatedAt.String()) } +type reliableDataInfo struct { + joiningMessageLock sync.Mutex + joiningMessageFirstSeqs map[livekit.ParticipantID]uint32 + joiningMessageLastWrittenSeqs map[livekit.ParticipantID]uint32 + lastPubReliableSeq atomic.Uint32 + stopReliableByMigrateOut atomic.Bool + canWriteReliable bool + migrateInPubDataCache atomic.Pointer[MigrationDataCache] +} + // --------------------------------------------------------------- type ParticipantParams struct { @@ -168,6 +179,7 @@ type ParticipantParams struct { DatachannelSlowThreshold int FireOnTrackBySdp bool DisableCodecRegression bool + LastPubReliableSeq uint32 } type ParticipantImpl struct { @@ -233,6 +245,8 @@ type ParticipantImpl struct { dataChannelStats *telemetry.BytesTrackStats + reliableDataInfo reliableDataInfo + rttUpdatedAt time.Time lastRTT uint32 @@ -302,6 +316,10 @@ func NewParticipant(params ParticipantParams) (*ParticipantImpl, error) { connectionQuality: livekit.ConnectionQuality_EXCELLENT, pubLogger: params.Logger.WithComponent(sutils.ComponentPub), subLogger: params.Logger.WithComponent(sutils.ComponentSub), + reliableDataInfo: reliableDataInfo{ + joiningMessageFirstSeqs: make(map[livekit.ParticipantID]uint32), + joiningMessageLastWrittenSeqs: make(map[livekit.ParticipantID]uint32), + }, } p.id.Store(params.SID) p.dataChannelStats = telemetry.NewBytesTrackStats( @@ -310,6 +328,7 @@ func NewParticipant(params ParticipantParams) (*ParticipantImpl, error) { params.Telemetry, params.Reporter, ) + p.reliableDataInfo.lastPubReliableSeq.Store(params.LastPubReliableSeq) p.participantHelper.Store(params.ParticipantHelper) if !params.DisableSupervisor { p.supervisor = supervisor.NewParticipantSupervisor(supervisor.ParticipantSupervisorParams{Logger: params.Logger}) @@ -1090,6 +1109,7 @@ func (p *ParticipantImpl) SetMigrateInfo( previousOffer, previousAnswer *webrtc.SessionDescription, mediaTracks []*livekit.TrackPublishedResponse, dataChannels []*livekit.DataChannelInfo, + dataChannelReceiveState []*livekit.DataChannelReceiveState, ) { p.pendingTracksLock.Lock() for _, t := range mediaTracks { @@ -1109,6 +1129,12 @@ func (p *ParticipantImpl) SetMigrateInfo( p.setIsPublisher(true) } + p.reliableDataInfo.joiningMessageLock.Lock() + for _, state := range dataChannelReceiveState { + p.reliableDataInfo.joiningMessageFirstSeqs[livekit.ParticipantID(state.PublisherSid)] = state.LastSeq + 1 + } + p.reliableDataInfo.joiningMessageLock.Unlock() + p.TransportManager.SetMigrateInfo(previousOffer, previousAnswer, dataChannels) } @@ -1306,6 +1332,10 @@ func (p *ParticipantImpl) SetMigrateState(s types.MigrateState) { case types.MigrateStateComplete: if preState == types.MigrateStateSync { p.params.Logger.Infow("migration complete") + + if p.params.LastPubReliableSeq > 0 { + p.reliableDataInfo.migrateInPubDataCache.Store(NewMigrationDataCache(p.params.LastPubReliableSeq, time.Now().Add(migrationWaitContinuousMsgDuration))) + } } p.TransportManager.ProcessPendingPublisherDataChannels() go p.cacheForwarderState() @@ -1824,6 +1854,10 @@ func (p *ParticipantImpl) updateState(state livekit.ParticipantInfo_State) { } } + if state == livekit.ParticipantInfo_ACTIVE { + p.replayJoiningReliableMessages() + } + p.params.Logger.Debugw("updating participant state", "state", state.String()) p.dirty.Store(true) @@ -1997,6 +2031,55 @@ func (p *ParticipantImpl) onReceivedDataMessage(kind livekit.DataPacket_Kind, da return } + dp.ParticipantSid = string(p.ID()) + if kind == livekit.DataPacket_RELIABLE && dp.Sequence > 0 { + if p.reliableDataInfo.stopReliableByMigrateOut.Load() { + return + } + + if migrationCache := p.reliableDataInfo.migrateInPubDataCache.Load(); migrationCache != nil { + switch migrationCache.Add(dp) { + case MigrationDataCacheStateWaiting: + // waiting for the reliable sequence to continue from last node + return + + case MigrationDataCacheStateTimeout: + p.reliableDataInfo.migrateInPubDataCache.Store(nil) + // waiting time out, handle all cached messages + cachedMsgs := migrationCache.Get() + if len(cachedMsgs) == 0 { + p.pubLogger.Warnw("migration data cache timed out, no cached messages received", nil, "lastPubReliableSeq", p.params.LastPubReliableSeq) + } else { + p.pubLogger.Warnw("migration data cache timed out, handling cached messages", nil, + "cachedFirstSeq", cachedMsgs[0].Sequence, + "cachedLastSeq", cachedMsgs[len(cachedMsgs)-1].Sequence, + "lastPubReliableSeq", p.params.LastPubReliableSeq, + ) + } + for _, cachedDp := range cachedMsgs { + p.handleReceivedDataMessage(kind, cachedDp) + } + return + + case MigrationDataCacheStateDone: + // see the continous message, drop the cache + p.reliableDataInfo.migrateInPubDataCache.Store(nil) + } + } + } + + p.handleReceivedDataMessage(kind, dp) +} + +func (p *ParticipantImpl) handleReceivedDataMessage(kind livekit.DataPacket_Kind, dp *livekit.DataPacket) { + if kind == livekit.DataPacket_RELIABLE && dp.Sequence > 0 { + if p.reliableDataInfo.lastPubReliableSeq.Load() >= dp.Sequence { + return + } + + p.reliableDataInfo.lastPubReliableSeq.Store(dp.Sequence) + } + // trust the channel that it came in as the source of truth dp.Kind = kind @@ -3173,11 +3256,36 @@ func (p *ParticipantImpl) SupportsTransceiverReuse() bool { return p.ProtocolVersion().SupportsTransceiverReuse() && !p.SupportsSyncStreamID() } -func (p *ParticipantImpl) SendDataMessage(kind livekit.DataPacket_Kind, data []byte) error { - if p.State() != livekit.ParticipantInfo_ACTIVE { - return ErrDataChannelUnavailable +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 p.State() != livekit.ParticipantInfo_ACTIVE { + return ErrDataChannelUnavailable + } + return p.TransportManager.SendDataMessage(kind, data) } + p.reliableDataInfo.joiningMessageLock.Lock() + if !p.reliableDataInfo.canWriteReliable { + if _, ok := p.reliableDataInfo.joiningMessageFirstSeqs[sender]; !ok { + p.reliableDataInfo.joiningMessageFirstSeqs[sender] = seq + } + p.reliableDataInfo.joiningMessageLock.Unlock() + return nil + } + + lastWrittenSeq, ok := p.reliableDataInfo.joiningMessageLastWrittenSeqs[sender] + if ok { + if seq <= lastWrittenSeq { + // already sent by replayJoiningReliableMessages + p.reliableDataInfo.joiningMessageLock.Unlock() + return nil + } else { + delete(p.reliableDataInfo.joiningMessageLastWrittenSeqs, sender) + } + } + + p.reliableDataInfo.joiningMessageLock.Unlock() + return p.TransportManager.SendDataMessage(kind, data) } @@ -3246,6 +3354,24 @@ func (p *ParticipantImpl) setupEnabledCodecs(publishEnabledCodecs []*livekit.Cod p.params.Logger.Debugw("setup enabled codecs", "publish", p.enabledPublishCodecs, "subscribe", p.enabledSubscribeCodecs, "disabled", disabledCodecs) } +func (p *ParticipantImpl) replayJoiningReliableMessages() { + p.reliableDataInfo.joiningMessageLock.Lock() + for _, msgCache := range p.helper().GetCachedReliableDataMessage(p.reliableDataInfo.joiningMessageFirstSeqs) { + if len(msgCache.DestIdentities) != 0 && !slices.Contains(msgCache.DestIdentities, p.Identity()) { + continue + } + if lastSeq, ok := p.reliableDataInfo.joiningMessageLastWrittenSeqs[msgCache.SenderID]; !ok || lastSeq < msgCache.Seq { + p.reliableDataInfo.joiningMessageLastWrittenSeqs[msgCache.SenderID] = msgCache.Seq + } + + p.TransportManager.SendDataMessage(livekit.DataPacket_RELIABLE, msgCache.Data) + } + + p.reliableDataInfo.joiningMessageFirstSeqs = make(map[livekit.ParticipantID]uint32) + p.reliableDataInfo.canWriteReliable = true + p.reliableDataInfo.joiningMessageLock.Unlock() +} + func (p *ParticipantImpl) GetEnabledPublishCodecs() []*livekit.Codec { codecs := make([]*livekit.Codec, 0, len(p.enabledPublishCodecs)) for _, c := range p.enabledPublishCodecs { @@ -3387,3 +3513,10 @@ func (p *ParticipantImpl) MoveToRoom(params types.MoveToRoomParams) { func (p *ParticipantImpl) helper() types.LocalParticipantHelper { return p.participantHelper.Load().(types.LocalParticipantHelper) } + +func (p *ParticipantImpl) GetLastReliableSequence(migrateOut bool) uint32 { + if migrateOut { + p.reliableDataInfo.stopReliableByMigrateOut.Store(true) + } + return p.reliableDataInfo.lastPubReliableSeq.Load() +} diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go index 5269fc48d..3e8942f7e 100644 --- a/pkg/rtc/room.go +++ b/pkg/rtc/room.go @@ -56,6 +56,9 @@ const ( dataForwardLoadBalanceThreshold = 4 simulateDisconnectSignalTimeout = 5 * time.Second + + dataMessageCacheTTL = 2 * time.Second + dataMessageCacheSize = 100_000 ) var ( @@ -139,6 +142,8 @@ type Room struct { disconnectSignalOnResumeNoMessagesParticipants map[livekit.ParticipantIdentity]*disconnectSignalOnResumeNoMessages userPacketDeduper *UserPacketDeduper + + dataMessagecache *utils.TimeSizeCache[types.DataMessageCache] } type ParticipantOptions struct { @@ -268,6 +273,10 @@ func NewRoom( disconnectSignalOnResumeParticipants: make(map[livekit.ParticipantIdentity]time.Time), disconnectSignalOnResumeNoMessagesParticipants: make(map[livekit.ParticipantIdentity]*disconnectSignalOnResumeNoMessages), userPacketDeduper: NewUserPacketDeduper(), + dataMessagecache: utils.NewTimeSizeCache[types.DataMessageCache](utils.TimeSizeCacheParams{ + TTL: dataMessageCacheTTL, + MaxSize: dataMessageCacheSize, + }), } if r.protoRoom.EmptyTimeout == 0 { @@ -649,6 +658,7 @@ func (r *Room) ResumeParticipant( IceServers: iceServers, ClientConfiguration: p.GetClientConfiguration(), ServerInfo: r.serverInfo, + LastMessageSeq: p.GetLastReliableSequence(false), }); err != nil { return err } @@ -1291,6 +1301,19 @@ func (r *Room) onParticipantUpdate(p types.LocalParticipant) { } func (r *Room) onDataPacket(source types.LocalParticipant, kind livekit.DataPacket_Kind, dp *livekit.DataPacket) { + if kind == livekit.DataPacket_RELIABLE && source != nil && dp.GetSequence() > 0 { + data, err := proto.Marshal(dp) + if err != nil { + r.logger.Errorw("failed to marshal data packet for cache", err, "participant", source.Identity(), "seq", dp.GetSequence()) + return + } + r.dataMessagecache.Add(&types.DataMessageCache{ + SenderID: source.ID(), + Seq: dp.GetSequence(), + Data: data, + DestIdentities: livekit.StringsAsIDs[livekit.ParticipantIdentity](dp.DestinationIdentities), + }, len(data)) + } BroadcastDataPacketForRoom(r, source, kind, dp, r.logger) } @@ -1400,6 +1423,8 @@ func (r *Room) changeUpdateWorker() { subTicker := time.NewTicker(subscriberUpdateInterval) defer subTicker.Stop() + cleanDataMessageTicker := time.NewTicker(dataMessageCacheTTL) + for !r.IsClosed() { select { case <-r.closed: @@ -1420,6 +1445,9 @@ func (r *Room) changeUpdateWorker() { r.batchedUpdatesMu.Unlock() SendParticipantUpdates(maps.Values(updatesMap), r.GetParticipants()) + + case <-cleanDataMessageTicker.C: + r.dataMessagecache.Prune() } } } @@ -1693,6 +1721,17 @@ func (r *Room) IsDataMessageUserPacketDuplicate(up *livekit.UserPacket) bool { return r.userPacketDeduper.IsDuplicate(up) } +func (r *Room) GetCachedReliableDataMessage(seqs map[livekit.ParticipantID]uint32) []*types.DataMessageCache { + msgs := make([]*types.DataMessageCache, 0, len(seqs)*10) + for _, msg := range r.dataMessagecache.Get() { + seq, ok := seqs[msg.SenderID] + if ok && msg.Seq >= seq { + msgs = append(msgs, msg) + } + } + return msgs +} + // ------------------------------------------------------------ func BroadcastDataPacketForRoom( @@ -1754,7 +1793,7 @@ func BroadcastDataPacketForRoom( } utils.ParallelExec(destParticipants, dataForwardLoadBalanceThreshold, 1, func(op types.LocalParticipant) { - op.SendDataMessage(kind, dpData) + op.SendDataMessage(kind, dpData, livekit.ParticipantID(dp.GetParticipantSid()), dp.GetSequence()) }) } diff --git a/pkg/rtc/room_test.go b/pkg/rtc/room_test.go index ff552f599..092f8a246 100644 --- a/pkg/rtc/room_test.go +++ b/pkg/rtc/room_test.go @@ -651,7 +651,7 @@ func TestDataChannel(t *testing.T) { continue } require.Equal(t, 1, fp.SendDataMessageCallCount()) - _, got := fp.SendDataMessageArgsForCall(0) + _, got, _, _ := fp.SendDataMessageArgsForCall(0) require.Equal(t, encoded, got) } }) @@ -698,7 +698,7 @@ func TestDataChannel(t *testing.T) { } } require.Equal(t, 1, p1.SendDataMessageCallCount()) - _, got := p1.SendDataMessageArgsForCall(0) + _, got, _, _ := p1.SendDataMessageArgsForCall(0) require.Equal(t, encoded, got) }) } diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index c9b909e6c..7117733f7 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -324,6 +324,13 @@ type MoveToRoomParams struct { Helper LocalParticipantHelper } +type DataMessageCache struct { + Data []byte + SenderID livekit.ParticipantID + Seq uint32 + DestIdentities []livekit.ParticipantIdentity +} + //counterfeiter:generate . LocalParticipantHelper type LocalParticipantHelper interface { ResolveMediaTrack(LocalParticipant, livekit.TrackID) MediaResolverResult @@ -331,6 +338,7 @@ type LocalParticipantHelper interface { GetRegionSettings(ip string) *livekit.RegionSettings GetSubscriberForwarderState(p LocalParticipant) (map[livekit.TrackID]*livekit.RTPForwarderState, error) ShouldRegressCodec() bool + GetCachedReliableDataMessage(seqs map[livekit.ParticipantID]uint32) []*DataMessageCache } //counterfeiter:generate . LocalParticipant @@ -365,6 +373,7 @@ type LocalParticipant interface { GetEnabledPublishCodecs() []*livekit.Codec GetPublisherICESessionUfrag() (string, error) SupportsMoving() bool + GetLastReliableSequence(migrateOut bool) uint32 SetResponseSink(sink routing.MessageSink) CloseSignalConnection(reason SignallingCloseReason) @@ -430,7 +439,7 @@ type LocalParticipant interface { SendJoinResponse(joinResponse *livekit.JoinResponse) error SendParticipantUpdate(participants []*livekit.ParticipantInfo) error SendSpeakerUpdate(speakers []*livekit.SpeakerInfo, force bool) error - SendDataMessage(kind livekit.DataPacket_Kind, data []byte) error + SendDataMessage(kind livekit.DataPacket_Kind, data []byte, senderID livekit.ParticipantID, seq uint32) error SendDataMessageUnlabeled(data []byte, useRaw bool, sender livekit.ParticipantIdentity) error SendRoomUpdate(room *livekit.Room) error SendConnectionQualityUpdate(update *livekit.ConnectionQualityUpdate) error @@ -470,6 +479,7 @@ type LocalParticipant interface { previousOffer, previousAnswer *webrtc.SessionDescription, mediaTracks []*livekit.TrackPublishedResponse, dataChannels []*livekit.DataChannelInfo, + dataChannelReceiveState []*livekit.DataChannelReceiveState, ) IsReconnect() bool MoveToRoom(params MoveToRoomParams) diff --git a/pkg/rtc/types/typesfakes/fake_local_participant.go b/pkg/rtc/types/typesfakes/fake_local_participant.go index 3126dc4c1..6a81a10a4 100644 --- a/pkg/rtc/types/typesfakes/fake_local_participant.go +++ b/pkg/rtc/types/typesfakes/fake_local_participant.go @@ -329,6 +329,17 @@ type FakeLocalParticipant struct { getICEConnectionInfoReturnsOnCall map[int]struct { result1 []*types.ICEConnectionInfo } + GetLastReliableSequenceStub func(bool) uint32 + getLastReliableSequenceMutex sync.RWMutex + getLastReliableSequenceArgsForCall []struct { + arg1 bool + } + getLastReliableSequenceReturns struct { + result1 uint32 + } + getLastReliableSequenceReturnsOnCall map[int]struct { + result1 uint32 + } GetLoggerStub func() logger.Logger getLoggerMutex sync.RWMutex getLoggerArgsForCall []struct { @@ -867,11 +878,13 @@ type FakeLocalParticipant struct { sendConnectionQualityUpdateReturnsOnCall map[int]struct { result1 error } - SendDataMessageStub func(livekit.DataPacket_Kind, []byte) error + SendDataMessageStub func(livekit.DataPacket_Kind, []byte, livekit.ParticipantID, uint32) error sendDataMessageMutex sync.RWMutex sendDataMessageArgsForCall []struct { arg1 livekit.DataPacket_Kind arg2 []byte + arg3 livekit.ParticipantID + arg4 uint32 } sendDataMessageReturns struct { result1 error @@ -985,13 +998,14 @@ type FakeLocalParticipant struct { setMetadataArgsForCall []struct { arg1 string } - SetMigrateInfoStub func(*webrtc.SessionDescription, *webrtc.SessionDescription, []*livekit.TrackPublishedResponse, []*livekit.DataChannelInfo) + SetMigrateInfoStub func(*webrtc.SessionDescription, *webrtc.SessionDescription, []*livekit.TrackPublishedResponse, []*livekit.DataChannelInfo, []*livekit.DataChannelReceiveState) setMigrateInfoMutex sync.RWMutex setMigrateInfoArgsForCall []struct { arg1 *webrtc.SessionDescription arg2 *webrtc.SessionDescription arg3 []*livekit.TrackPublishedResponse arg4 []*livekit.DataChannelInfo + arg5 []*livekit.DataChannelReceiveState } SetMigrateStateStub func(types.MigrateState) setMigrateStateMutex sync.RWMutex @@ -2890,6 +2904,67 @@ func (fake *FakeLocalParticipant) GetICEConnectionInfoReturnsOnCall(i int, resul }{result1} } +func (fake *FakeLocalParticipant) GetLastReliableSequence(arg1 bool) uint32 { + fake.getLastReliableSequenceMutex.Lock() + ret, specificReturn := fake.getLastReliableSequenceReturnsOnCall[len(fake.getLastReliableSequenceArgsForCall)] + fake.getLastReliableSequenceArgsForCall = append(fake.getLastReliableSequenceArgsForCall, struct { + arg1 bool + }{arg1}) + stub := fake.GetLastReliableSequenceStub + fakeReturns := fake.getLastReliableSequenceReturns + fake.recordInvocation("GetLastReliableSequence", []interface{}{arg1}) + fake.getLastReliableSequenceMutex.Unlock() + if stub != nil { + return stub(arg1) + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeLocalParticipant) GetLastReliableSequenceCallCount() int { + fake.getLastReliableSequenceMutex.RLock() + defer fake.getLastReliableSequenceMutex.RUnlock() + return len(fake.getLastReliableSequenceArgsForCall) +} + +func (fake *FakeLocalParticipant) GetLastReliableSequenceCalls(stub func(bool) uint32) { + fake.getLastReliableSequenceMutex.Lock() + defer fake.getLastReliableSequenceMutex.Unlock() + fake.GetLastReliableSequenceStub = stub +} + +func (fake *FakeLocalParticipant) GetLastReliableSequenceArgsForCall(i int) bool { + fake.getLastReliableSequenceMutex.RLock() + defer fake.getLastReliableSequenceMutex.RUnlock() + argsForCall := fake.getLastReliableSequenceArgsForCall[i] + return argsForCall.arg1 +} + +func (fake *FakeLocalParticipant) GetLastReliableSequenceReturns(result1 uint32) { + fake.getLastReliableSequenceMutex.Lock() + defer fake.getLastReliableSequenceMutex.Unlock() + fake.GetLastReliableSequenceStub = nil + fake.getLastReliableSequenceReturns = struct { + result1 uint32 + }{result1} +} + +func (fake *FakeLocalParticipant) GetLastReliableSequenceReturnsOnCall(i int, result1 uint32) { + fake.getLastReliableSequenceMutex.Lock() + defer fake.getLastReliableSequenceMutex.Unlock() + fake.GetLastReliableSequenceStub = nil + if fake.getLastReliableSequenceReturnsOnCall == nil { + fake.getLastReliableSequenceReturnsOnCall = make(map[int]struct { + result1 uint32 + }) + } + fake.getLastReliableSequenceReturnsOnCall[i] = struct { + result1 uint32 + }{result1} +} + func (fake *FakeLocalParticipant) GetLogger() logger.Logger { fake.getLoggerMutex.Lock() ret, specificReturn := fake.getLoggerReturnsOnCall[len(fake.getLoggerArgsForCall)] @@ -5848,7 +5923,7 @@ func (fake *FakeLocalParticipant) SendConnectionQualityUpdateReturnsOnCall(i int }{result1} } -func (fake *FakeLocalParticipant) SendDataMessage(arg1 livekit.DataPacket_Kind, arg2 []byte) error { +func (fake *FakeLocalParticipant) SendDataMessage(arg1 livekit.DataPacket_Kind, arg2 []byte, arg3 livekit.ParticipantID, arg4 uint32) error { var arg2Copy []byte if arg2 != nil { arg2Copy = make([]byte, len(arg2)) @@ -5859,13 +5934,15 @@ func (fake *FakeLocalParticipant) SendDataMessage(arg1 livekit.DataPacket_Kind, fake.sendDataMessageArgsForCall = append(fake.sendDataMessageArgsForCall, struct { arg1 livekit.DataPacket_Kind arg2 []byte - }{arg1, arg2Copy}) + arg3 livekit.ParticipantID + arg4 uint32 + }{arg1, arg2Copy, arg3, arg4}) stub := fake.SendDataMessageStub fakeReturns := fake.sendDataMessageReturns - fake.recordInvocation("SendDataMessage", []interface{}{arg1, arg2Copy}) + fake.recordInvocation("SendDataMessage", []interface{}{arg1, arg2Copy, arg3, arg4}) fake.sendDataMessageMutex.Unlock() if stub != nil { - return stub(arg1, arg2) + return stub(arg1, arg2, arg3, arg4) } if specificReturn { return ret.result1 @@ -5879,17 +5956,17 @@ func (fake *FakeLocalParticipant) SendDataMessageCallCount() int { return len(fake.sendDataMessageArgsForCall) } -func (fake *FakeLocalParticipant) SendDataMessageCalls(stub func(livekit.DataPacket_Kind, []byte) error) { +func (fake *FakeLocalParticipant) SendDataMessageCalls(stub func(livekit.DataPacket_Kind, []byte, livekit.ParticipantID, uint32) error) { fake.sendDataMessageMutex.Lock() defer fake.sendDataMessageMutex.Unlock() fake.SendDataMessageStub = stub } -func (fake *FakeLocalParticipant) SendDataMessageArgsForCall(i int) (livekit.DataPacket_Kind, []byte) { +func (fake *FakeLocalParticipant) SendDataMessageArgsForCall(i int) (livekit.DataPacket_Kind, []byte, livekit.ParticipantID, uint32) { fake.sendDataMessageMutex.RLock() defer fake.sendDataMessageMutex.RUnlock() argsForCall := fake.sendDataMessageArgsForCall[i] - return argsForCall.arg1, argsForCall.arg2 + return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3, argsForCall.arg4 } func (fake *FakeLocalParticipant) SendDataMessageReturns(result1 error) { @@ -6517,7 +6594,7 @@ func (fake *FakeLocalParticipant) SetMetadataArgsForCall(i int) string { return argsForCall.arg1 } -func (fake *FakeLocalParticipant) SetMigrateInfo(arg1 *webrtc.SessionDescription, arg2 *webrtc.SessionDescription, arg3 []*livekit.TrackPublishedResponse, arg4 []*livekit.DataChannelInfo) { +func (fake *FakeLocalParticipant) SetMigrateInfo(arg1 *webrtc.SessionDescription, arg2 *webrtc.SessionDescription, arg3 []*livekit.TrackPublishedResponse, arg4 []*livekit.DataChannelInfo, arg5 []*livekit.DataChannelReceiveState) { var arg3Copy []*livekit.TrackPublishedResponse if arg3 != nil { arg3Copy = make([]*livekit.TrackPublishedResponse, len(arg3)) @@ -6528,18 +6605,24 @@ func (fake *FakeLocalParticipant) SetMigrateInfo(arg1 *webrtc.SessionDescription arg4Copy = make([]*livekit.DataChannelInfo, len(arg4)) copy(arg4Copy, arg4) } + var arg5Copy []*livekit.DataChannelReceiveState + if arg5 != nil { + arg5Copy = make([]*livekit.DataChannelReceiveState, len(arg5)) + copy(arg5Copy, arg5) + } fake.setMigrateInfoMutex.Lock() fake.setMigrateInfoArgsForCall = append(fake.setMigrateInfoArgsForCall, struct { arg1 *webrtc.SessionDescription arg2 *webrtc.SessionDescription arg3 []*livekit.TrackPublishedResponse arg4 []*livekit.DataChannelInfo - }{arg1, arg2, arg3Copy, arg4Copy}) + arg5 []*livekit.DataChannelReceiveState + }{arg1, arg2, arg3Copy, arg4Copy, arg5Copy}) stub := fake.SetMigrateInfoStub - fake.recordInvocation("SetMigrateInfo", []interface{}{arg1, arg2, arg3Copy, arg4Copy}) + fake.recordInvocation("SetMigrateInfo", []interface{}{arg1, arg2, arg3Copy, arg4Copy, arg5Copy}) fake.setMigrateInfoMutex.Unlock() if stub != nil { - fake.SetMigrateInfoStub(arg1, arg2, arg3, arg4) + fake.SetMigrateInfoStub(arg1, arg2, arg3, arg4, arg5) } } @@ -6549,17 +6632,17 @@ func (fake *FakeLocalParticipant) SetMigrateInfoCallCount() int { return len(fake.setMigrateInfoArgsForCall) } -func (fake *FakeLocalParticipant) SetMigrateInfoCalls(stub func(*webrtc.SessionDescription, *webrtc.SessionDescription, []*livekit.TrackPublishedResponse, []*livekit.DataChannelInfo)) { +func (fake *FakeLocalParticipant) SetMigrateInfoCalls(stub func(*webrtc.SessionDescription, *webrtc.SessionDescription, []*livekit.TrackPublishedResponse, []*livekit.DataChannelInfo, []*livekit.DataChannelReceiveState)) { fake.setMigrateInfoMutex.Lock() defer fake.setMigrateInfoMutex.Unlock() fake.SetMigrateInfoStub = stub } -func (fake *FakeLocalParticipant) SetMigrateInfoArgsForCall(i int) (*webrtc.SessionDescription, *webrtc.SessionDescription, []*livekit.TrackPublishedResponse, []*livekit.DataChannelInfo) { +func (fake *FakeLocalParticipant) SetMigrateInfoArgsForCall(i int) (*webrtc.SessionDescription, *webrtc.SessionDescription, []*livekit.TrackPublishedResponse, []*livekit.DataChannelInfo, []*livekit.DataChannelReceiveState) { fake.setMigrateInfoMutex.RLock() defer fake.setMigrateInfoMutex.RUnlock() argsForCall := fake.setMigrateInfoArgsForCall[i] - return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3, argsForCall.arg4 + return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3, argsForCall.arg4, argsForCall.arg5 } func (fake *FakeLocalParticipant) SetMigrateState(arg1 types.MigrateState) { @@ -8310,6 +8393,8 @@ func (fake *FakeLocalParticipant) Invocations() map[string][][]interface{} { defer fake.getICEConfigMutex.RUnlock() fake.getICEConnectionInfoMutex.RLock() defer fake.getICEConnectionInfoMutex.RUnlock() + fake.getLastReliableSequenceMutex.RLock() + defer fake.getLastReliableSequenceMutex.RUnlock() fake.getLoggerMutex.RLock() defer fake.getLoggerMutex.RUnlock() fake.getLoggerResolverMutex.RLock() diff --git a/pkg/rtc/types/typesfakes/fake_local_participant_helper.go b/pkg/rtc/types/typesfakes/fake_local_participant_helper.go index 6e7d8ea77..00df1c304 100644 --- a/pkg/rtc/types/typesfakes/fake_local_participant_helper.go +++ b/pkg/rtc/types/typesfakes/fake_local_participant_helper.go @@ -9,6 +9,17 @@ import ( ) type FakeLocalParticipantHelper struct { + GetCachedReliableDataMessageStub func(map[livekit.ParticipantID]uint32) []*types.DataMessageCache + getCachedReliableDataMessageMutex sync.RWMutex + getCachedReliableDataMessageArgsForCall []struct { + arg1 map[livekit.ParticipantID]uint32 + } + getCachedReliableDataMessageReturns struct { + result1 []*types.DataMessageCache + } + getCachedReliableDataMessageReturnsOnCall map[int]struct { + result1 []*types.DataMessageCache + } GetParticipantInfoStub func(livekit.ParticipantID) *livekit.ParticipantInfo getParticipantInfoMutex sync.RWMutex getParticipantInfoArgsForCall []struct { @@ -70,6 +81,67 @@ type FakeLocalParticipantHelper struct { invocationsMutex sync.RWMutex } +func (fake *FakeLocalParticipantHelper) GetCachedReliableDataMessage(arg1 map[livekit.ParticipantID]uint32) []*types.DataMessageCache { + fake.getCachedReliableDataMessageMutex.Lock() + ret, specificReturn := fake.getCachedReliableDataMessageReturnsOnCall[len(fake.getCachedReliableDataMessageArgsForCall)] + fake.getCachedReliableDataMessageArgsForCall = append(fake.getCachedReliableDataMessageArgsForCall, struct { + arg1 map[livekit.ParticipantID]uint32 + }{arg1}) + stub := fake.GetCachedReliableDataMessageStub + fakeReturns := fake.getCachedReliableDataMessageReturns + fake.recordInvocation("GetCachedReliableDataMessage", []interface{}{arg1}) + fake.getCachedReliableDataMessageMutex.Unlock() + if stub != nil { + return stub(arg1) + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeLocalParticipantHelper) GetCachedReliableDataMessageCallCount() int { + fake.getCachedReliableDataMessageMutex.RLock() + defer fake.getCachedReliableDataMessageMutex.RUnlock() + return len(fake.getCachedReliableDataMessageArgsForCall) +} + +func (fake *FakeLocalParticipantHelper) GetCachedReliableDataMessageCalls(stub func(map[livekit.ParticipantID]uint32) []*types.DataMessageCache) { + fake.getCachedReliableDataMessageMutex.Lock() + defer fake.getCachedReliableDataMessageMutex.Unlock() + fake.GetCachedReliableDataMessageStub = stub +} + +func (fake *FakeLocalParticipantHelper) GetCachedReliableDataMessageArgsForCall(i int) map[livekit.ParticipantID]uint32 { + fake.getCachedReliableDataMessageMutex.RLock() + defer fake.getCachedReliableDataMessageMutex.RUnlock() + argsForCall := fake.getCachedReliableDataMessageArgsForCall[i] + return argsForCall.arg1 +} + +func (fake *FakeLocalParticipantHelper) GetCachedReliableDataMessageReturns(result1 []*types.DataMessageCache) { + fake.getCachedReliableDataMessageMutex.Lock() + defer fake.getCachedReliableDataMessageMutex.Unlock() + fake.GetCachedReliableDataMessageStub = nil + fake.getCachedReliableDataMessageReturns = struct { + result1 []*types.DataMessageCache + }{result1} +} + +func (fake *FakeLocalParticipantHelper) GetCachedReliableDataMessageReturnsOnCall(i int, result1 []*types.DataMessageCache) { + fake.getCachedReliableDataMessageMutex.Lock() + defer fake.getCachedReliableDataMessageMutex.Unlock() + fake.GetCachedReliableDataMessageStub = nil + if fake.getCachedReliableDataMessageReturnsOnCall == nil { + fake.getCachedReliableDataMessageReturnsOnCall = make(map[int]struct { + result1 []*types.DataMessageCache + }) + } + fake.getCachedReliableDataMessageReturnsOnCall[i] = struct { + result1 []*types.DataMessageCache + }{result1} +} + func (fake *FakeLocalParticipantHelper) GetParticipantInfo(arg1 livekit.ParticipantID) *livekit.ParticipantInfo { fake.getParticipantInfoMutex.Lock() ret, specificReturn := fake.getParticipantInfoReturnsOnCall[len(fake.getParticipantInfoArgsForCall)] @@ -374,6 +446,8 @@ func (fake *FakeLocalParticipantHelper) ShouldRegressCodecReturnsOnCall(i int, r func (fake *FakeLocalParticipantHelper) Invocations() map[string][][]interface{} { fake.invocationsMutex.RLock() defer fake.invocationsMutex.RUnlock() + fake.getCachedReliableDataMessageMutex.RLock() + defer fake.getCachedReliableDataMessageMutex.RUnlock() fake.getParticipantInfoMutex.RLock() defer fake.getParticipantInfoMutex.RUnlock() fake.getRegionSettingsMutex.RLock() diff --git a/pkg/service/roommanager.go b/pkg/service/roommanager.go index ad49c20b9..0a384dd0c 100644 --- a/pkg/service/roommanager.go +++ b/pkg/service/roommanager.go @@ -1069,3 +1069,7 @@ func (h *roomManagerParticipantHelper) ResolveMediaTrack(lp types.LocalParticipa func (h *roomManagerParticipantHelper) ShouldRegressCodec() bool { return h.codecRegressionThreshold == 0 || h.room.GetParticipantCount() < h.codecRegressionThreshold } + +func (h *roomManagerParticipantHelper) GetCachedReliableDataMessage(seqs map[livekit.ParticipantID]uint32) []*types.DataMessageCache { + return h.room.GetCachedReliableDataMessage(seqs) +}