webrtc: sort tracks in a deterministic way (#5988) (#5989)

When ingesting tracks with WebRTC, track order was randomized,
preventing multi-track always-available streams from working reliably,
since they require tracks to be ordered in a precise way.

WebRTC tracks are not ordered by MID, RID, trackID and streamID
respectively.

internal/protocols/webrtc/inbound_track.go

@@ -249,7 +249,9 @@ type InboundTrack struct {
 
 	track     *webrtc.TrackRemote
 	receiver  *webrtc.RTPReceiver
+	midIndex  int
 	rid       string
+	ridIndex  int
 	writeRTCP func([]rtcp.Packet) error
 	log       logger.Writer
 

internal/protocols/webrtc/peer_connection.go

@@ -80,6 +80,25 @@ func maxTrackCount(medias []*sdp.MediaDescription) int {
 	return total
 }
 
+func ridIndexByMedia(media *sdp.MediaDescription) map[string]int {
+	ret := make(map[string]int)
+
+	for _, attr := range media.Attributes {
+		if attr.Key != "rid" || attr.Value == "" {
+			continue
+		}
+
+		ridge, _, ok := strings.Cut(attr.Value, " ")
+		if !ok || ridge == "" {
+			ridge = attr.Value
+		}
+
+		ret[ridge] = len(ret)
+	}
+
+	return ret
+}
+
 // * skip ConfigureRTCPReports
 // * add statsInterceptor
 func registerInterceptors(
@@ -767,31 +786,70 @@ func (co *PeerConnection) GatherInboundTracks(timeout time.Duration) error {
 	sdp.Unmarshal([]byte(co.wr.RemoteDescription().SDP)) //nolint:errcheck
 
 	maxTrackCount := maxTrackCount(sdp.MediaDescriptions)
+	tracks := make([]*InboundTrack, 0, maxTrackCount)
 
 	t := time.NewTimer(timeout)
 	defer t.Stop()
 
+	midIndexByMid := make(map[string]int, len(sdp.MediaDescriptions))
+	ridIndexByMid := make(map[string]map[string]int, len(sdp.MediaDescriptions))
+
+	for i, media := range sdp.MediaDescriptions {
+		mid, _ := media.Attribute("mid")
+		midIndexByMid[mid] = i
+		ridIndexByMid[mid] = ridIndexByMedia(media)
+	}
+
+	getMIDIndex := func(mid string) int {
+		if v, ok := midIndexByMid[mid]; ok {
+			return v
+		}
+
+		return len(midIndexByMid)
+	}
+
+	getRIDIndex := func(mid, rid string) int {
+		if rid == "" {
+			return 0
+		}
+
+		if v, ok := ridIndexByMid[mid][rid]; ok {
+			return v
+		}
+
+		return len(ridIndexByMid[mid])
+	}
+
+outer:
 	for {
 		select {
 		case <-t.C:
-			if len(co.inboundTracks) != 0 {
-				return nil
+			if len(tracks) != 0 {
+				break outer
 			}
 			return fmt.Errorf("deadline exceeded while waiting tracks")
 
 		case pair := <-co.inboundTrack:
-			t := &InboundTrack{
+			mid := ""
+			if transceiver := pair.receiver.RTPTransceiver(); transceiver != nil {
+				mid = transceiver.Mid()
+			}
+			rid := pair.track.RID()
+
+			track := &InboundTrack{
 				track:     pair.track,
 				receiver:  pair.receiver,
-				rid:       pair.track.RID(),
+				midIndex:  getMIDIndex(mid),
+				rid:       rid,
+				ridIndex:  getRIDIndex(mid, rid),
 				writeRTCP: co.wr.WriteRTCP,
 				log:       co.Log,
 			}
-			t.initialize()
-			co.inboundTracks = append(co.inboundTracks, t)
+			track.initialize()
+			tracks = append(tracks, track)
 
-			if len(co.inboundTracks) >= maxTrackCount {
-				return nil
+			if len(tracks) >= maxTrackCount {
+				break outer
 			}
 
 		case <-co.Failed():
@@ -801,6 +859,29 @@ func (co *PeerConnection) GatherInboundTracks(timeout time.Duration) error {
 			return fmt.Errorf("terminated")
 		}
 	}
+
+	slices.SortStableFunc(tracks, func(track1 *InboundTrack, track2 *InboundTrack) int {
+		if track1.midIndex != track2.midIndex {
+			return track1.midIndex - track2.midIndex
+		}
+
+		if track1.ridIndex != track2.ridIndex {
+			return track1.ridIndex - track2.ridIndex
+		}
+
+		id1 := track1.track.ID()
+		id2 := track2.track.ID()
+		if id1 != id2 {
+			return strings.Compare(id1, id2)
+		}
+
+		streamID1 := track1.track.StreamID()
+		streamID2 := track2.track.StreamID()
+		return strings.Compare(streamID1, streamID2)
+	})
+
+	co.inboundTracks = tracks
+	return nil
 }
 
 // Connected returns when connected.

internal/protocols/webrtc/peer_connection_test.go

@@ -34,6 +34,38 @@ func gatherCodecs(tracks []*InboundTrack) []webrtc.RTPCodecParameters {
 	return codecs
 }
 
+func gatherTrackIDs(tracks []*InboundTrack) []string {
+	ids := make([]string, len(tracks))
+	for i, track := range tracks {
+		ids[i] = track.track.ID()
+	}
+	return ids
+}
+
+func gatherTrackRIDs(tracks []*InboundTrack) []string {
+	rids := make([]string, len(tracks))
+	for i, track := range tracks {
+		rids[i] = track.track.RID()
+	}
+	return rids
+}
+
+func gatherTrackMIDIndexes(tracks []*InboundTrack) []int {
+	mids := make([]int, len(tracks))
+	for i, track := range tracks {
+		mids[i] = track.midIndex
+	}
+	return mids
+}
+
+func gatherTrackRIDIndexes(tracks []*InboundTrack) []int {
+	rids := make([]int, len(tracks))
+	for i, track := range tracks {
+		rids[i] = track.ridIndex
+	}
+	return rids
+}
+
 func senderHeaderExtensionID(params webrtc.RTPSendParameters, uri string) uint8 {
 	for _, ext := range params.HeaderExtensions {
 		if ext.URI == uri {
@@ -343,7 +375,7 @@ func TestPeerConnectionRead(t *testing.T) {
 	go func() {
 		time.Sleep(100 * time.Millisecond)
 
-		err2 := videoTrack.WriteRTP(&rtp.Packet{
+		err2 := audioTrack.WriteRTP(&rtp.Packet{
 			Header: rtp.Header{
 				Version:        2,
 				Marker:         true,
@@ -356,7 +388,7 @@ func TestPeerConnectionRead(t *testing.T) {
 		})
 		require.NoError(t, err2)
 
-		err2 = audioTrack.WriteRTP(&rtp.Packet{
+		err2 = videoTrack.WriteRTP(&rtp.Packet{
 			Header: rtp.Header{
 				Version:        2,
 				Marker:         true,
@@ -373,6 +405,9 @@ func TestPeerConnectionRead(t *testing.T) {
 	err = reader.GatherInboundTracks(2 * time.Second)
 	require.NoError(t, err)
 
+	require.Equal(t, []string{"video", "audio"}, gatherTrackIDs(reader.InboundTracks()))
+	require.Equal(t, []int{0, 1}, gatherTrackMIDIndexes(reader.InboundTracks()))
+
 	codecs := gatherCodecs(reader.InboundTracks())
 
 	sort.Slice(codecs, func(i, j int) bool {
@@ -572,12 +607,8 @@ func TestPeerConnectionReadSimulcast(t *testing.T) {
 		}, codec.RTPCodecCapability)
 	}
 
-	rids := make([]string, len(tracks))
-	for i, track := range tracks {
-		rids[i] = track.track.RID()
-	}
-	sort.Strings(rids)
-	require.Equal(t, []string{"h", "l", "m"}, rids)
+	require.Equal(t, []string{"l", "m", "h"}, gatherTrackRIDs(tracks))
+	require.Equal(t, []int{0, 1, 2}, gatherTrackRIDIndexes(tracks))
 }
 
 func TestPeerConnectionStripIncomingTWCC(t *testing.T) {