e2e reliability for data channel (#3716)

* e2e reliability for data channel

* solve comments

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

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=

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

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

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

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

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

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)

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

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

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