conference: fix the subscribe/unsubscribe logic

Author: daniel-abramov

pkg/conference/matrix.go

@@ -21,7 +21,6 @@ func (c *Conference) onNewParticipant(participantID ParticipantID, inviteEvent *
 	logger := c.logger.WithFields(logrus.Fields{
 		"user_id":   participantID.UserID,
 		"device_id": participantID.DeviceID,
-		"call_id":   participantID.CallID,
 	})
 
 	logger.Info("Incoming call invite")

pkg/conference/processor.go

@@ -7,6 +7,7 @@ import (
 	"github.com/matrix-org/waterfall/pkg/common"
 	"github.com/matrix-org/waterfall/pkg/peer"
 	"github.com/pion/webrtc/v3"
+	"golang.org/x/exp/slices"
 	"maunium.net/go/mautrix/event"
 )
 
@@ -85,7 +86,7 @@ func (c *Conference) processPeerMessage(message common.Message[ParticipantID, pe
 		}
 
 	case peer.NewICECandidate:
-		participant.logger.Info("Received a new local ICE candidate")
+		participant.logger.Debug("Received a new local ICE candidate")
 
 		// Convert WebRTC ICE candidate to Matrix ICE candidate.
 		jsonCandidate := msg.Candidate.ToJSON()
@@ -111,7 +112,7 @@ func (c *Conference) processPeerMessage(message common.Message[ParticipantID, pe
 		})
 
 	case peer.DataChannelMessage:
-		participant.logger.Info("Sent data channel message")
+		participant.logger.Debug("Received data channel message")
 		var sfuMessage event.SFUMessage
 		if err := json.Unmarshal([]byte(msg.Message), &sfuMessage); err != nil {
 			c.logger.Errorf("Failed to unmarshal SFU message: %v", err)
@@ -136,30 +137,44 @@ func (c *Conference) processPeerMessage(message common.Message[ParticipantID, pe
 func (c *Conference) handleDataChannelMessage(participant *Participant, sfuMessage event.SFUMessage) {
 	switch sfuMessage.Op {
 	case event.SFUOperationSelect:
-		participant.logger.Info("Sent select request over DC")
-
-		// Get the tracks that correspond to the tracks that the participant wants to receive.
-		for _, track := range c.getTracks(sfuMessage.Start) {
-			if track == nil {
-				participant.logger.Errorf("Bug, track is nil")
+		participant.logger.Info("Received select request over DC")
+
+		// Find tracks based on what we were asked for.
+		tracks := c.getTracks(sfuMessage.Start)
+
+		// Let's check if we have all the tracks that we were asked for are there.
+		// If not, we will list which are not available (later on we must inform participant
+		// about it unless the participant retries it).
+		if len(tracks) != len(sfuMessage.Start) {
+			for _, expected := range sfuMessage.Start {
+				found := slices.IndexFunc(tracks, func(track *webrtc.TrackLocalStaticRTP) bool {
+					return track.StreamID() == expected.StreamID && track.ID() == expected.TrackID
+				})
+
+				if found == -1 {
+					c.logger.Warnf("Track not found: %s", expected.TrackID)
+				}
 			}
+		}
 
+		// Subscribe to the found tracks.
+		for _, track := range tracks {
 			if err := participant.peer.SubscribeTo(track); err != nil {
 				participant.logger.Errorf("Failed to subscribe to track: %v", err)
 				return
 			}
 		}
 
 	case event.SFUOperationAnswer:
-		participant.logger.Info("Sent SDP answer over DC")
+		participant.logger.Info("Received SDP answer over DC")
 
 		if err := participant.peer.ProcessSDPAnswer(sfuMessage.SDP); err != nil {
 			participant.logger.Errorf("Failed to set SDP answer: %v", err)
 			return
 		}
 
 	case event.SFUOperationPublish:
-		participant.logger.Info("Sent SDP offer over DC")
+		participant.logger.Info("Received SDP offer over DC")
 
 		answer, err := participant.peer.ProcessSDPOffer(sfuMessage.SDP)
 		if err != nil {
@@ -173,13 +188,13 @@ func (c *Conference) handleDataChannelMessage(participant *Participant, sfuMessa
 		})
 
 	case event.SFUOperationUnpublish:
-		participant.logger.Info("Sent unpublish over DC")
+		participant.logger.Info("Received unpublish over DC")
 
 		// TODO: Clarify the semantics of unpublish.
 	case event.SFUOperationAlive:
 		// FIXME: Handle the heartbeat message here (updating the last timestamp etc).
 	case event.SFUOperationMetadata:
-		participant.logger.Info("Sent metadata over DC")
+		participant.logger.Info("Received metadata over DC")
 
 		participant.streamMetadata = sfuMessage.Metadata
 		c.resendMetadataToAllExcept(participant.id)

pkg/conference/state.go

@@ -88,6 +88,7 @@ func (c *Conference) getTracks(identifiers []event.SFUTrackDescription) []*webrt
 			}
 		}
 	}
+
 	return tracks
 }
 

pkg/peer/peer.go

@@ -106,6 +106,10 @@ func (p *Peer[ID]) SubscribeTo(track *webrtc.TrackLocalStaticRTP) error {
 func (p *Peer[ID]) UnsubscribeFrom(tracks []*webrtc.TrackLocalStaticRTP) {
 	// That's unfortunately an O(m*n) operation, but we don't expect the number of tracks to be big.
 	for _, presentTrack := range p.peerConnection.GetSenders() {
+		if presentTrack.Track() == nil {
+			continue
+		}
+
 		for _, trackToUnsubscribe := range tracks {
 			presentTrackID, presentStreamID := presentTrack.Track().ID(), presentTrack.Track().StreamID()
 			trackID, streamID := trackToUnsubscribe.ID(), trackToUnsubscribe.StreamID()
@@ -165,7 +169,6 @@ func (p *Peer[ID]) ProcessSDPAnswer(sdpAnswer string) error {
 
 // Applies the sdp offer received from the remote peer and generates an SDP answer.
 func (p *Peer[ID]) ProcessSDPOffer(sdpOffer string) (*webrtc.SessionDescription, error) {
-	p.logger.WithField("sdpOffer", sdpOffer).Debug("processing SDP offer")
 	err := p.peerConnection.SetRemoteDescription(webrtc.SessionDescription{
 		Type: webrtc.SDPTypeOffer,
 		SDP:  sdpOffer,

pkg/peer/webrtc.go

@@ -23,6 +23,7 @@ func (p *Peer[ID]) onRtpTrackReceived(remoteTrack *webrtc.TrackRemote, receiver
 			rtcp := []rtcp.Packet{&rtcp.PictureLossIndication{MediaSSRC: uint32(remoteTrack.SSRC())}}
 			if rtcpSendErr := p.peerConnection.WriteRTCP(rtcp); rtcpSendErr != nil {
 				p.logger.Errorf("Failed to send RTCP PLI: %v", rtcpSendErr)
+				return
 			}
 		}
 	}()
@@ -56,12 +57,14 @@ func (p *Peer[ID]) onRtpTrackReceived(remoteTrack *webrtc.TrackRemote, receiver
 					p.logger.WithError(readErr).Error("failed to read from remote track")
 				}
 				p.sink.Send(PublishedTrackFailed{Track: localTrack})
+				return
 			}
 
 			// ErrClosedPipe means we don't have any subscribers, this is ok if no peers have connected yet.
 			if _, err = localTrack.Write(rtpBuf[:index]); err != nil && !errors.Is(err, io.ErrClosedPipe) {
 				p.logger.WithError(err).Error("failed to write to local track")
 				p.sink.Send(PublishedTrackFailed{Track: localTrack})
+				return
 			}
 		}
 	}()
@@ -98,7 +101,7 @@ func (p *Peer[ID]) onNegotiationNeeded() {
 
 // A callback that is called once we receive an ICE connection state change for this peer connection.
 func (p *Peer[ID]) onICEConnectionStateChanged(state webrtc.ICEConnectionState) {
-	p.logger.WithField("state", state).Debug("ICE connection state changed")
+	p.logger.WithField("state", state).Info("ICE connection state changed")
 
 	// TODO: Ask Simon if we should do it here as in the previous implementation.
 	switch state {
@@ -120,7 +123,7 @@ func (p *Peer[ID]) onSignalingStateChanged(state webrtc.SignalingState) {
 }
 
 func (p *Peer[ID]) onConnectionStateChanged(state webrtc.PeerConnectionState) {
-	p.logger.WithField("state", state).Debug("connection state changed")
+	p.logger.WithField("state", state).Info("Connection state changed")
 
 	switch state {
 	case webrtc.PeerConnectionStateFailed, webrtc.PeerConnectionStateDisconnected, webrtc.PeerConnectionStateClosed:
@@ -136,33 +139,33 @@ func (p *Peer[ID]) onDataChannelReady(dc *webrtc.DataChannel) {
 	defer p.dataChannelMutex.Unlock()
 
 	if p.dataChannel != nil {
-		p.logger.Error("data channel already exists")
+		p.logger.Error("Data channel already exists")
 		p.dataChannel.Close()
 		return
 	}
 
 	p.dataChannel = dc
-	p.logger.WithField("label", dc.Label()).Info("data channel ready")
+	p.logger.WithField("label", dc.Label()).Info("Data channel ready")
 
 	dc.OnOpen(func() {
-		p.logger.Info("data channel opened")
+		p.logger.Info("Data channel opened")
 		p.sink.Send(DataChannelAvailable{})
 	})
 
 	dc.OnMessage(func(msg webrtc.DataChannelMessage) {
-		p.logger.WithField("message", msg).Debug("data channel message received")
+		p.logger.WithField("message", msg).Debug("Data channel message received")
 		if msg.IsString {
 			p.sink.Send(DataChannelMessage{Message: string(msg.Data)})
 		} else {
-			p.logger.Warn("data channel message is not a string, ignoring")
+			p.logger.Warn("Data channel message is not a string, ignoring")
 		}
 	})
 
 	dc.OnError(func(err error) {
-		p.logger.WithError(err).Error("data channel error")
+		p.logger.WithError(err).Error("Data channel error")
 	})
 
 	dc.OnClose(func() {
-		p.logger.Info("data channel closed")
+		p.logger.Info("Data channel closed")
 	})
 }

pkg/signaling/matrix.go

@@ -17,8 +17,6 @@ limitations under the License.
 package signaling
 
 import (
-	"encoding/json"
-
 	"github.com/sirupsen/logrus"
 	"maunium.net/go/mautrix"
 	"maunium.net/go/mautrix/event"
@@ -144,16 +142,6 @@ func (m *MatrixForConference) sendToDevice(user MatrixRecipient, eventType event
 		},
 	}
 
-	{
-		// TODO: Remove this once
-		serialized, err := json.Marshal(sendRequest)
-		if err != nil {
-			logger.WithError(err).Error("Failed to serialize to-device message")
-			return
-		}
-		logger.Debugf("Sending to-device message: %s", string(serialized))
-	}
-
 	if _, err := m.client.SendToDevice(eventType, sendRequest); err != nil {
 		logger.Errorf("failed to send to-device event: %w", err)
 	}