peer: implement heartbeat handling for keepalive

Note that we start the keepalive routine once the peer is created. We do
it like this since the keepalive deadline is actually quite high and I
would say that if within that deadline no heartbeat messages were sent,
then we can consider the connection as stalled.

I.e. starting the keepalive timer only once the peer is connected is
like sparing a second that a peer normally needs to establish a
connection?

Author: daniel-abramov

pkg/conference/config.go

@@ -3,6 +3,6 @@ package conference
 // Configuration for the group conferences (calls).
 type Config struct {
 	// Keep-alive timeout for WebRTC connections. If no keep-alive has been received
-	// from the client for this duration, the connection is considered dead.
+	// from the client for this duration, the connection is considered dead (in seconds).
 	KeepAliveTimeout int `yaml:"timeout"`
 }

pkg/conference/matrix.go

@@ -1,6 +1,8 @@
 package conference
 
 import (
+	"time"
+
 	"github.com/matrix-org/waterfall/pkg/common"
 	"github.com/matrix-org/waterfall/pkg/peer"
 	"github.com/pion/webrtc/v3"
@@ -50,7 +52,8 @@ func (c *Conference) onNewParticipant(participantID ParticipantID, inviteEvent *
 	} else {
 		messageSink := common.NewMessageSink(participantID, c.peerMessages)
 
-		peer, answer, err := peer.NewPeer(inviteEvent.Offer.SDP, messageSink, logger)
+		keepAliveDeadline := time.Duration(c.config.KeepAliveTimeout) * time.Second
+		peer, answer, err := peer.NewPeer(inviteEvent.Offer.SDP, messageSink, logger, keepAliveDeadline)
 		if err != nil {
 			logger.WithError(err).Errorf("Failed to process SDP offer")
 			return err

pkg/conference/processor.go

@@ -51,9 +51,9 @@ func (c *Conference) processPeerMessage(message common.Message[ParticipantID, pe
 		participant.logger.Info("Joined the call")
 
 	case peer.LeftTheCall:
-		participant.logger.Info("Left the call")
+		participant.logger.Info("Left the call: %s", msg.Reason)
 		c.removeParticipant(message.Sender)
-		c.signaling.SendHangup(participant.asMatrixRecipient(), event.CallHangupUnknownError)
+		c.signaling.SendHangup(participant.asMatrixRecipient(), msg.Reason)
 
 	case peer.NewTrackPublished:
 		participant.logger.Infof("Published new track: %s", msg.Track.ID())
@@ -195,9 +195,9 @@ func (c *Conference) handleDataChannelMessage(participant *Participant, sfuMessa
 	case event.SFUOperationUnpublish:
 		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).
+		participant.peer.ProcessHeartbeat()
+
 	case event.SFUOperationMetadata:
 		participant.streamMetadata = sfuMessage.Metadata
 		c.resendMetadataToAllExcept(participant.id)

pkg/peer/keepalive.go

@@ -0,0 +1,23 @@
+package peer
+
+import "time"
+
+type HeartBeat struct{}
+
+// Starts a goroutine that will execute `onDeadLine` closure in case nothing has been published
+// to the `heartBeat` channel for `deadline` duration. The goroutine stops once the channel is closed.
+func startKeepAlive(deadline time.Duration, heartBeat <-chan HeartBeat, onDeadLine func()) {
+	go func() {
+		for {
+			select {
+			case <-time.After(deadline):
+				onDeadLine()
+				return
+			case _, ok := <-heartBeat:
+				if !ok {
+					return
+				}
+			}
+		}
+	}()
+}

pkg/peer/messages.go

@@ -2,6 +2,7 @@ package peer
 
 import (
 	"github.com/pion/webrtc/v3"
+	"maunium.net/go/mautrix/event"
 )
 
 // Due to the limitation of Go, we're using the `interface{}` to be able to use switch the actual
@@ -10,7 +11,9 @@ type MessageContent = interface{}
 
 type JoinedTheCall struct{}
 
-type LeftTheCall struct{}
+type LeftTheCall struct {
+	Reason event.CallHangupReason
+}
 
 type NewTrackPublished struct {
 	Track *webrtc.TrackLocalStaticRTP

pkg/peer/peer.go

@@ -3,10 +3,12 @@ package peer
 import (
 	"errors"
 	"sync"
+	"time"
 
 	"github.com/matrix-org/waterfall/pkg/common"
 	"github.com/pion/webrtc/v3"
 	"github.com/sirupsen/logrus"
+	"maunium.net/go/mautrix/event"
 )
 
 var (
@@ -28,6 +30,7 @@ type Peer[ID comparable] struct {
 	logger         *logrus.Entry
 	peerConnection *webrtc.PeerConnection
 	sink           *common.MessageSink[ID, MessageContent]
+	heartbeat      chan HeartBeat
 
 	dataChannelMutex sync.Mutex
 	dataChannel      *webrtc.DataChannel
@@ -38,6 +41,7 @@ func NewPeer[ID comparable](
 	sdpOffer string,
 	sink *common.MessageSink[ID, MessageContent],
 	logger *logrus.Entry,
+	keepAliveDeadline time.Duration,
 ) (*Peer[ID], *webrtc.SessionDescription, error) {
 	peerConnection, err := webrtc.NewPeerConnection(webrtc.Configuration{})
 	if err != nil {
@@ -49,6 +53,7 @@ func NewPeer[ID comparable](
 		logger:         logger,
 		peerConnection: peerConnection,
 		sink:           sink,
+		heartbeat:      make(chan HeartBeat, common.UnboundedChannelSize),
 	}
 
 	peerConnection.OnTrack(peer.onRtpTrackReceived)
@@ -63,6 +68,8 @@ func NewPeer[ID comparable](
 	if sdpAnswer, err := peer.ProcessSDPOffer(sdpOffer); err != nil {
 		return nil, nil, err
 	} else {
+		onDeadline := func() { peer.sink.Send(LeftTheCall{event.CallHangupKeepAliveTimeout}) }
+		startKeepAlive(keepAliveDeadline, peer.heartbeat, onDeadline)
 		return peer, sdpAnswer, nil
 	}
 }
@@ -192,3 +199,10 @@ func (p *Peer[ID]) ProcessSDPOffer(sdpOffer string) (*webrtc.SessionDescription,
 
 	return &answer, nil
 }
+
+// New heartbeat received (keep-alive message that is periodically sent by the remote peer).
+// We need to update the last heartbeat time. If the peer is not active for too long, we will
+// consider peer's connection as stalled and will close it.
+func (p *Peer[ID]) ProcessHeartbeat() {
+	p.heartbeat <- HeartBeat{}
+}

pkg/peer/webrtc.go

@@ -7,6 +7,7 @@ import (
 
 	"github.com/pion/rtcp"
 	"github.com/pion/webrtc/v3"
+	"maunium.net/go/mautrix/event"
 )
 
 // A callback that is called once we receive first RTP packets from a track, i.e.
@@ -127,7 +128,7 @@ func (p *Peer[ID]) onConnectionStateChanged(state webrtc.PeerConnectionState) {
 
 	switch state {
 	case webrtc.PeerConnectionStateFailed, webrtc.PeerConnectionStateDisconnected, webrtc.PeerConnectionStateClosed:
-		p.sink.Send(LeftTheCall{})
+		p.sink.Send(LeftTheCall{event.CallHangupUserHangup})
 	case webrtc.PeerConnectionStateConnected:
 		p.sink.Send(JoinedTheCall{})
 	}