Simplify migrate complete a bit more. (#2491)

Moving handling of migrated tracks to when the migration state moves to
completed. Pending data channel were already happening only on complete.
Move tracks also to that point.

Handling it earlier meant that track published callback happened and
ownership of track moved to new node before the new node could finish
peer connection. So, in cases where migration did not go through, this
caused confusion of track ownership.

pkg/rtc/participant.go

@@ -202,8 +202,8 @@ type ParticipantImpl struct {
 
 	lock utils.RWMutex
 
-	dirty        atomic.Bool
-	version      atomic.Uint32
+	dirty   atomic.Bool
+	version atomic.Uint32
 
 	// callbacks & handlers
 	onTrackPublished     func(types.LocalParticipant, types.MediaTrack)
@@ -654,18 +654,11 @@ func (p *ParticipantImpl) onPublisherAnswer(answer webrtc.SessionDescription) er
 
 	p.pubLogger.Debugw("sending answer", "transport", livekit.SignalTarget_PUBLISHER)
 	answer = p.configurePublisherAnswer(answer)
-	if err := p.writeMessage(&livekit.SignalResponse{
+	return p.writeMessage(&livekit.SignalResponse{
 		Message: &livekit.SignalResponse_Answer{
 			Answer: ToProtoSessionDescription(answer),
 		},
-	}); err != nil {
-		return err
-	}
-
-	if p.MigrateState() == types.MigrateStateSync {
-		go p.handleMigrateTracks()
-	}
-	return nil
+	})
 }
 
 func (p *ParticipantImpl) handleMigrateTracks() {
@@ -910,6 +903,7 @@ func (p *ParticipantImpl) SetMigrateState(s types.MigrateState) {
 		p.TransportManager.ProcessPendingPublisherOffer()
 
 	case types.MigrateStateComplete:
+		p.handleMigrateTracks()
 		p.TransportManager.ProcessPendingPublisherDataChannels()
 	}
 
@@ -1482,9 +1476,7 @@ func (p *ParticipantImpl) onICECandidate(c *webrtc.ICECandidate, target livekit.
 }
 
 func (p *ParticipantImpl) onPublisherInitialConnected() {
-	if !p.hasPendingMigratedTrack() {
-		p.SetMigrateState(types.MigrateStateComplete)
-	}
+	p.SetMigrateState(types.MigrateStateComplete)
 
 	if p.supervisor != nil {
 		p.supervisor.SetPublisherPeerConnectionConnected(true)
@@ -1533,8 +1525,7 @@ func (p *ParticipantImpl) setupDisconnectTimer() {
 		if p.IsClosed() || p.IsDisconnected() {
 			return
 		}
-		reason := types.ParticipantCloseReasonPeerConnectionDisconnected
-		_ = p.Close(true, reason, false)
+		_ = p.Close(true, types.ParticipantCloseReasonPeerConnectionDisconnected, false)
 	})
 	p.lock.Unlock()
 }
@@ -2106,10 +2097,6 @@ func (p *ParticipantImpl) handleTrackPublished(track types.MediaTrack) {
 	p.pendingTracksLock.Lock()
 	delete(p.pendingPublishingTracks, track.ID())
 	p.pendingTracksLock.Unlock()
-
-	if !p.hasPendingMigratedTrack() && p.TransportManager.HasPublisherEverConnected() {
-		p.SetMigrateState(types.MigrateStateComplete)
-	}
 }
 
 func (p *ParticipantImpl) hasPendingMigratedTrack() bool {

pkg/rtc/participant_signal.go

@@ -100,7 +100,13 @@ func (p *ParticipantImpl) SendParticipantUpdate(participantsToUpdate []*livekit.
 			// this is a message delivered out of order, a more recent version of the message had already been
 			// sent.
 			if pi.Version < lastVersion.version {
-				p.params.Logger.Debugw("skipping outdated participant update", "otherParticipant", pi.Identity, "otherPID", pi.Sid, "version", pi.Version, "lastVersion", lastVersion)
+				p.params.Logger.Debugw(
+					"skipping outdated participant update",
+					"otherParticipant", pi.Identity,
+					"otherPID", pi.Sid,
+					"version", pi.Version,
+					"lastVersion", lastVersion,
+				)
 				isValid = false
 			}
 		}