fix bugs with websocket reconnection, don't re-use redis sinks (#22)

cmd/cli/client/client.go

@@ -127,19 +127,12 @@ func NewRTCClient(conn *websocket.Conn) (*RTCClient, error) {
 	})
 
 	c.subscriber.PeerConnection().OnTrack(func(track *webrtc.TrackRemote, rtpReceiver *webrtc.RTPReceiver) {
-		logger.Debugw("track received", "label", track.StreamID(), "id", track.ID(),
-			"participant", c.localParticipant.Identity)
 		go c.processTrack(track)
 	})
 	c.subscriber.PeerConnection().OnDataChannel(func(channel *webrtc.DataChannel) {
 	})
 
-	c.publisher.OnNegotiationNeeded(func() {
-		if !c.iceConnected.Get() {
-			return
-		}
-		c.negotiate()
-	})
+	c.publisher.OnNegotiationNeeded(c.negotiate)
 
 	c.publisher.PeerConnection().OnICEConnectionStateChange(func(connectionState webrtc.ICEConnectionState) {
 		logger.Debugw("ICE state has changed", "state", connectionState.String(),
@@ -329,7 +322,14 @@ func (c *RTCClient) ReadResponse() (*livekit.SignalResponse, error) {
 
 // TODO: this function is not thread safe, need to cleanup
 func (c *RTCClient) SubscribedTracks() map[string][]*webrtc.TrackRemote {
-	return c.subscribedTracks
+	// create a copy of this
+	c.lock.Lock()
+	defer c.lock.Unlock()
+	tracks := make(map[string][]*webrtc.TrackRemote, len(c.subscribedTracks))
+	for key, val := range c.subscribedTracks {
+		tracks[key] = val
+	}
+	return tracks
 }
 
 func (c *RTCClient) RemoteParticipants() []*livekit.ParticipantInfo {
@@ -433,7 +433,7 @@ func (c *RTCClient) AddFileTrack(path string, id string, label string) (writer *
 		return nil, fmt.Errorf("%s has an unsupported extension", filepath.Base(path))
 	}
 
-	logger.Debugw("adding track",
+	logger.Debugw("adding file track",
 		"mime", mime,
 	)
 
@@ -506,18 +506,18 @@ func (c *RTCClient) handleAnswer(desc webrtc.SessionDescription) error {
 	return nil
 }
 
-func (c *RTCClient) negotiate() error {
+func (c *RTCClient) negotiate() {
 	logger.Debugw("starting negotiation", "participant", c.localParticipant.Identity)
 	offer, err := c.publisher.PeerConnection().CreateOffer(nil)
 	if err != nil {
-		return err
+		return
 	}
 
 	if err := c.publisher.PeerConnection().SetLocalDescription(offer); err != nil {
-		return err
+		return
 	}
 
-	return c.SendRequest(&livekit.SignalRequest{
+	c.SendRequest(&livekit.SignalRequest{
 		Message: &livekit.SignalRequest_Offer{
 			Offer: rtc.ToProtoSessionDescription(offer),
 		},
@@ -526,11 +526,17 @@ func (c *RTCClient) negotiate() error {
 
 func (c *RTCClient) processTrack(track *webrtc.TrackRemote) {
 	lastUpdate := time.Time{}
-	pId, trackId := rtc.UnpackTrackId(track.ID())
+	pId := track.StreamID()
+	trackId := track.ID()
 	c.lock.Lock()
 	c.subscribedTracks[pId] = append(c.subscribedTracks[pId], track)
 	c.lock.Unlock()
 
+	logger.Debugw("client added track", "participant", c.localParticipant.Identity,
+		"source", pId,
+		"track", trackId,
+	)
+
 	defer func() {
 		c.lock.Lock()
 		c.subscribedTracks[pId] = funk.Without(c.subscribedTracks[pId], track).([]*webrtc.TrackRemote)

pkg/routing/interfaces.go

@@ -39,9 +39,10 @@ type Router interface {
 	ListNodes() ([]*livekit.Node, error)
 
 	// participant signal connection is ready to start
-	StartParticipantSignal(roomName, identity, metadata string, reconnect bool) (reqSink MessageSink, resSource MessageSource, err error)
+	StartParticipantSignal(roomName, identity, metadata string, reconnect bool) (connectionId string, reqSink MessageSink, resSource MessageSource, err error)
+
 	// sends a message to RTC node
-	SendRTCMessage(roomName, identity string, msg *livekit.RTCNodeMessage) error
+	CreateRTCSink(roomName, identity string) (MessageSink, error)
 
 	// when a new participant's RTC connection is ready to start
 	OnNewParticipantRTC(callback NewParticipantCallback)

pkg/routing/localrouter.go

@@ -4,6 +4,8 @@ import (
 	"sync"
 	"time"
 
+	"github.com/livekit/livekit-server/pkg/logger"
+	"github.com/livekit/livekit-server/pkg/utils"
 	"github.com/livekit/livekit-server/proto/livekit"
 )
 
@@ -14,6 +16,9 @@ type LocalRouter struct {
 	// channels for each participant
 	requestChannels  map[string]*MessageChannel
 	responseChannels map[string]*MessageChannel
+	isStarted        utils.AtomicFlag
+
+	rtcMessageChan *MessageChannel
 
 	onNewParticipant NewParticipantCallback
 	onRTCMessage     RTCMessageCallback
@@ -24,6 +29,7 @@ func NewLocalRouter(currentNode LocalNode) *LocalRouter {
 		currentNode:      currentNode,
 		requestChannels:  make(map[string]*MessageChannel),
 		responseChannels: make(map[string]*MessageChannel),
+		rtcMessageChan:   NewMessageChannel(),
 	}
 }
 
@@ -65,10 +71,11 @@ func (r *LocalRouter) ListNodes() ([]*livekit.Node, error) {
 	}, nil
 }
 
-func (r *LocalRouter) StartParticipantSignal(roomName, identity, metadata string, reconnect bool) (reqSink MessageSink, resSource MessageSource, err error) {
+func (r *LocalRouter) StartParticipantSignal(roomName, identity, metadata string, reconnect bool) (connectionId string, reqSink MessageSink, resSource MessageSource, err error) {
 	// treat it as a new participant connecting
 	if r.onNewParticipant == nil {
-		return nil, nil, ErrHandlerNotDefined
+		err = ErrHandlerNotDefined
+		return
 	}
 
 	// index channels by roomName | identity
@@ -86,15 +93,15 @@ func (r *LocalRouter) StartParticipantSignal(roomName, identity, metadata string
 		// response sink
 		resChan,
 	)
-	return reqChan, resChan, nil
+	return identity, reqChan, resChan, nil
 }
 
-func (r *LocalRouter) SendRTCMessage(roomName, identity string, msg *livekit.RTCNodeMessage) error {
-	if r.onRTCMessage == nil {
-		return nil
+func (r *LocalRouter) CreateRTCSink(roomName, identity string) (MessageSink, error) {
+	if r.rtcMessageChan.isClosed.Get() {
+		// create a new one
+		r.rtcMessageChan = NewMessageChannel()
 	}
-	go r.onRTCMessage(roomName, identity, msg)
-	return nil
+	return r.rtcMessageChan, nil
 }
 
 func (r *LocalRouter) OnNewParticipantRTC(callback NewParticipantCallback) {
@@ -106,22 +113,62 @@ func (r *LocalRouter) OnRTCMessage(callback RTCMessageCallback) {
 }
 
 func (r *LocalRouter) Start() error {
+	if !r.isStarted.TrySet(true) {
+		return nil
+	}
 	go r.statsWorker()
 	// on local routers, Start doesn't do anything, websocket connections initiate the connections
+	go r.rtcMessageWorker()
 	return nil
 }
 
 func (r *LocalRouter) Stop() {
+
+	r.rtcMessageChan.Close()
 }
 
 func (r *LocalRouter) statsWorker() {
 	for {
+		if !r.isStarted.Get() {
+			return
+		}
 		// update every 10 seconds
 		<-time.After(statsUpdateInterval)
 		r.currentNode.Stats.UpdatedAt = time.Now().Unix()
 	}
 }
 
+func (r *LocalRouter) rtcMessageWorker() {
+	// is a new channel available? if so swap to that one
+	if !r.isStarted.Get() {
+		return
+	}
+
+	// start a new worker after this finished
+	defer func() {
+		go r.rtcMessageWorker()
+	}()
+
+	if r.rtcMessageChan.isClosed.Get() {
+		// sleep and retry
+		time.Sleep(time.Second)
+	}
+
+	// consume messages from
+	for msg := range r.rtcMessageChan.ReadChan() {
+		if rtcMsg, ok := msg.(*livekit.RTCNodeMessage); ok {
+			room, identity, err := parseParticipantKey(rtcMsg.ParticipantKey)
+			if err != nil {
+				logger.Errorw("could not process RTC message", "error", err)
+				continue
+			}
+			if r.onRTCMessage != nil {
+				r.onRTCMessage(room, identity, rtcMsg)
+			}
+		}
+	}
+}
+
 func (r *LocalRouter) getOrCreateMessageChannel(target map[string]*MessageChannel, key string) *MessageChannel {
 	r.lock.Lock()
 	defer r.lock.Unlock()

pkg/routing/redisrouter.go

@@ -28,8 +28,6 @@ type RedisRouter struct {
 	ctx       context.Context
 	isStarted utils.AtomicFlag
 
-	// map of participantKey => RTCNodeSink
-	rtcSinks map[string]*RTCNodeSink
 	// map of connectionId => SignalNodeSink
 	signalSinks map[string]*SignalNodeSink
 
@@ -41,7 +39,6 @@ func NewRedisRouter(currentNode LocalNode, rc *redis.Client) *RedisRouter {
 	rr := &RedisRouter{
 		LocalRouter: *NewLocalRouter(currentNode),
 		rc:          rc,
-		rtcSinks:    make(map[string]*RTCNodeSink),
 		signalSinks: make(map[string]*SignalNodeSink),
 	}
 	rr.ctx, rr.cancel = context.WithCancel(context.Background())
@@ -127,7 +124,7 @@ func (r *RedisRouter) ListNodes() ([]*livekit.Node, error) {
 }
 
 // signal connection sets up paths to the RTC node, and starts to route messages to that message queue
-func (r *RedisRouter) StartParticipantSignal(roomName, identity, metadata string, reconnect bool) (reqSink MessageSink, resSource MessageSource, err error) {
+func (r *RedisRouter) StartParticipantSignal(roomName, identity, metadata string, reconnect bool) (connectionId string, reqSink MessageSink, resSource MessageSource, err error) {
 	// find the node where the room is hosted at
 	rtcNode, err := r.GetNodeForRoom(roomName)
 	if err != nil {
@@ -135,7 +132,7 @@ func (r *RedisRouter) StartParticipantSignal(roomName, identity, metadata string
 	}
 
 	// create a new connection id
-	connectionId := utils.NewGuid("CO_")
+	connectionId = utils.NewGuid("CO_")
 	pKey := participantKey(roomName, identity)
 
 	// map signal & rtc nodes
@@ -143,7 +140,7 @@ func (r *RedisRouter) StartParticipantSignal(roomName, identity, metadata string
 		return
 	}
 
-	sink := r.getOrCreateRTCSink(rtcNode, pKey)
+	sink := NewRTCNodeSink(r.rc, rtcNode, pKey)
 
 	// sends a message to start session
 	err = sink.WriteMessage(&livekit.StartSession{
@@ -160,19 +157,17 @@ func (r *RedisRouter) StartParticipantSignal(roomName, identity, metadata string
 
 	// index by connectionId, since there may be multiple connections for the participant
 	resChan := r.getOrCreateMessageChannel(r.responseChannels, connectionId)
-	return sink, resChan, nil
+	return connectionId, sink, resChan, nil
 }
 
-func (r *RedisRouter) SendRTCMessage(roomName, identity string, msg *livekit.RTCNodeMessage) error {
+func (r *RedisRouter) CreateRTCSink(roomName, identity string) (MessageSink, error) {
 	pkey := participantKey(roomName, identity)
 	rtcNode, err := r.getParticipantRTCNode(pkey)
 	if err != nil {
-		return err
+		return nil, err
 	}
 
-	rtcSink := r.getOrCreateRTCSink(rtcNode, pkey)
-
-	return rtcSink.WriteMessage(msg)
+	return NewRTCNodeSink(r.rc, rtcNode, pkey), nil
 }
 
 func (r *RedisRouter) startParticipantRTC(ss *livekit.StartSession, participantKey string) error {
@@ -216,7 +211,7 @@ func (r *RedisRouter) startParticipantRTC(ss *livekit.StartSession, participantK
 	}
 
 	reqChan := r.getOrCreateMessageChannel(r.requestChannels, participantKey)
-	resSink := r.getOrCreateSignalSink(signalNode, ss.ConnectionId)
+	resSink := NewSignalNodeSink(r.rc, signalNode, ss.ConnectionId)
 	r.onNewParticipant(
 		ss.RoomName,
 		ss.Identity,
@@ -261,44 +256,6 @@ func (r *RedisRouter) setParticipantSignalNode(connectionId, nodeId string) erro
 	return nil
 }
 
-func (r *RedisRouter) getOrCreateRTCSink(nodeId string, participantKey string) *RTCNodeSink {
-	r.lock.Lock()
-	defer r.lock.Unlock()
-	sink := r.rtcSinks[participantKey]
-
-	if sink != nil {
-		return sink
-	}
-
-	sink = NewRTCNodeSink(r.rc, nodeId, participantKey)
-	sink.OnClose(func() {
-		r.lock.Lock()
-		delete(r.rtcSinks, participantKey)
-		r.lock.Unlock()
-	})
-	r.rtcSinks[participantKey] = sink
-	return sink
-}
-
-func (r *RedisRouter) getOrCreateSignalSink(nodeId string, connectionId string) *SignalNodeSink {
-	r.lock.Lock()
-	defer r.lock.Unlock()
-	sink := r.signalSinks[connectionId]
-
-	if sink != nil {
-		return sink
-	}
-
-	sink = NewSignalNodeSink(r.rc, nodeId, connectionId)
-	sink.OnClose(func() {
-		r.lock.Lock()
-		delete(r.signalSinks, connectionId)
-		r.lock.Unlock()
-	})
-	r.signalSinks[connectionId] = sink
-	return sink
-}
-
 func (r *RedisRouter) getParticipantRTCNode(participantKey string) (string, error) {
 	return r.rc.Get(r.ctx, participantRTCKey(participantKey)).Result()
 }

pkg/routing/routingfakes/fake_router.go

@@ -20,6 +20,20 @@ type FakeRouter struct {
 	clearRoomStateReturnsOnCall map[int]struct {
 		result1 error
 	}
+	CreateRTCSinkStub        func(string, string) (routing.MessageSink, error)
+	createRTCSinkMutex       sync.RWMutex
+	createRTCSinkArgsForCall []struct {
+		arg1 string
+		arg2 string
+	}
+	createRTCSinkReturns struct {
+		result1 routing.MessageSink
+		result2 error
+	}
+	createRTCSinkReturnsOnCall map[int]struct {
+		result1 routing.MessageSink
+		result2 error
+	}
 	GetNodeStub        func(string) (*livekit.Node, error)
 	getNodeMutex       sync.RWMutex
 	getNodeArgsForCall []struct {
@@ -88,19 +102,6 @@ type FakeRouter struct {
 	removeDeadNodesReturnsOnCall map[int]struct {
 		result1 error
 	}
-	SendRTCMessageStub        func(string, string, *livekit.RTCNodeMessage) error
-	sendRTCMessageMutex       sync.RWMutex
-	sendRTCMessageArgsForCall []struct {
-		arg1 string
-		arg2 string
-		arg3 *livekit.RTCNodeMessage
-	}
-	sendRTCMessageReturns struct {
-		result1 error
-	}
-	sendRTCMessageReturnsOnCall map[int]struct {
-		result1 error
-	}
 	SetNodeForRoomStub        func(string, string) error
 	setNodeForRoomMutex       sync.RWMutex
 	setNodeForRoomArgsForCall []struct {
@@ -123,7 +124,7 @@ type FakeRouter struct {
 	startReturnsOnCall map[int]struct {
 		result1 error
 	}
-	StartParticipantSignalStub        func(string, string, string, bool) (routing.MessageSink, routing.MessageSource, error)
+	StartParticipantSignalStub        func(string, string, string, bool) (string, routing.MessageSink, routing.MessageSource, error)
 	startParticipantSignalMutex       sync.RWMutex
 	startParticipantSignalArgsForCall []struct {
 		arg1 string
@@ -132,14 +133,16 @@ type FakeRouter struct {
 		arg4 bool
 	}
 	startParticipantSignalReturns struct {
-		result1 routing.MessageSink
-		result2 routing.MessageSource
-		result3 error
+		result1 string
+		result2 routing.MessageSink
+		result3 routing.MessageSource
+		result4 error
 	}
 	startParticipantSignalReturnsOnCall map[int]struct {
-		result1 routing.MessageSink
-		result2 routing.MessageSource
-		result3 error
+		result1 string
+		result2 routing.MessageSink
+		result3 routing.MessageSource
+		result4 error
 	}
 	StopStub        func()
 	stopMutex       sync.RWMutex
@@ -220,6 +223,71 @@ func (fake *FakeRouter) ClearRoomStateReturnsOnCall(i int, result1 error) {
 	}{result1}
 }
 
+func (fake *FakeRouter) CreateRTCSink(arg1 string, arg2 string) (routing.MessageSink, error) {
+	fake.createRTCSinkMutex.Lock()
+	ret, specificReturn := fake.createRTCSinkReturnsOnCall[len(fake.createRTCSinkArgsForCall)]
+	fake.createRTCSinkArgsForCall = append(fake.createRTCSinkArgsForCall, struct {
+		arg1 string
+		arg2 string
+	}{arg1, arg2})
+	stub := fake.CreateRTCSinkStub
+	fakeReturns := fake.createRTCSinkReturns
+	fake.recordInvocation("CreateRTCSink", []interface{}{arg1, arg2})
+	fake.createRTCSinkMutex.Unlock()
+	if stub != nil {
+		return stub(arg1, arg2)
+	}
+	if specificReturn {
+		return ret.result1, ret.result2
+	}
+	return fakeReturns.result1, fakeReturns.result2
+}
+
+func (fake *FakeRouter) CreateRTCSinkCallCount() int {
+	fake.createRTCSinkMutex.RLock()
+	defer fake.createRTCSinkMutex.RUnlock()
+	return len(fake.createRTCSinkArgsForCall)
+}
+
+func (fake *FakeRouter) CreateRTCSinkCalls(stub func(string, string) (routing.MessageSink, error)) {
+	fake.createRTCSinkMutex.Lock()
+	defer fake.createRTCSinkMutex.Unlock()
+	fake.CreateRTCSinkStub = stub
+}
+
+func (fake *FakeRouter) CreateRTCSinkArgsForCall(i int) (string, string) {
+	fake.createRTCSinkMutex.RLock()
+	defer fake.createRTCSinkMutex.RUnlock()
+	argsForCall := fake.createRTCSinkArgsForCall[i]
+	return argsForCall.arg1, argsForCall.arg2
+}
+
+func (fake *FakeRouter) CreateRTCSinkReturns(result1 routing.MessageSink, result2 error) {
+	fake.createRTCSinkMutex.Lock()
+	defer fake.createRTCSinkMutex.Unlock()
+	fake.CreateRTCSinkStub = nil
+	fake.createRTCSinkReturns = struct {
+		result1 routing.MessageSink
+		result2 error
+	}{result1, result2}
+}
+
+func (fake *FakeRouter) CreateRTCSinkReturnsOnCall(i int, result1 routing.MessageSink, result2 error) {
+	fake.createRTCSinkMutex.Lock()
+	defer fake.createRTCSinkMutex.Unlock()
+	fake.CreateRTCSinkStub = nil
+	if fake.createRTCSinkReturnsOnCall == nil {
+		fake.createRTCSinkReturnsOnCall = make(map[int]struct {
+			result1 routing.MessageSink
+			result2 error
+		})
+	}
+	fake.createRTCSinkReturnsOnCall[i] = struct {
+		result1 routing.MessageSink
+		result2 error
+	}{result1, result2}
+}
+
 func (fake *FakeRouter) GetNode(arg1 string) (*livekit.Node, error) {
 	fake.getNodeMutex.Lock()
 	ret, specificReturn := fake.getNodeReturnsOnCall[len(fake.getNodeArgsForCall)]
@@ -574,69 +642,6 @@ func (fake *FakeRouter) RemoveDeadNodesReturnsOnCall(i int, result1 error) {
 	}{result1}
 }
 
-func (fake *FakeRouter) SendRTCMessage(arg1 string, arg2 string, arg3 *livekit.RTCNodeMessage) error {
-	fake.sendRTCMessageMutex.Lock()
-	ret, specificReturn := fake.sendRTCMessageReturnsOnCall[len(fake.sendRTCMessageArgsForCall)]
-	fake.sendRTCMessageArgsForCall = append(fake.sendRTCMessageArgsForCall, struct {
-		arg1 string
-		arg2 string
-		arg3 *livekit.RTCNodeMessage
-	}{arg1, arg2, arg3})
-	stub := fake.SendRTCMessageStub
-	fakeReturns := fake.sendRTCMessageReturns
-	fake.recordInvocation("SendRTCMessage", []interface{}{arg1, arg2, arg3})
-	fake.sendRTCMessageMutex.Unlock()
-	if stub != nil {
-		return stub(arg1, arg2, arg3)
-	}
-	if specificReturn {
-		return ret.result1
-	}
-	return fakeReturns.result1
-}
-
-func (fake *FakeRouter) SendRTCMessageCallCount() int {
-	fake.sendRTCMessageMutex.RLock()
-	defer fake.sendRTCMessageMutex.RUnlock()
-	return len(fake.sendRTCMessageArgsForCall)
-}
-
-func (fake *FakeRouter) SendRTCMessageCalls(stub func(string, string, *livekit.RTCNodeMessage) error) {
-	fake.sendRTCMessageMutex.Lock()
-	defer fake.sendRTCMessageMutex.Unlock()
-	fake.SendRTCMessageStub = stub
-}
-
-func (fake *FakeRouter) SendRTCMessageArgsForCall(i int) (string, string, *livekit.RTCNodeMessage) {
-	fake.sendRTCMessageMutex.RLock()
-	defer fake.sendRTCMessageMutex.RUnlock()
-	argsForCall := fake.sendRTCMessageArgsForCall[i]
-	return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3
-}
-
-func (fake *FakeRouter) SendRTCMessageReturns(result1 error) {
-	fake.sendRTCMessageMutex.Lock()
-	defer fake.sendRTCMessageMutex.Unlock()
-	fake.SendRTCMessageStub = nil
-	fake.sendRTCMessageReturns = struct {
-		result1 error
-	}{result1}
-}
-
-func (fake *FakeRouter) SendRTCMessageReturnsOnCall(i int, result1 error) {
-	fake.sendRTCMessageMutex.Lock()
-	defer fake.sendRTCMessageMutex.Unlock()
-	fake.SendRTCMessageStub = nil
-	if fake.sendRTCMessageReturnsOnCall == nil {
-		fake.sendRTCMessageReturnsOnCall = make(map[int]struct {
-			result1 error
-		})
-	}
-	fake.sendRTCMessageReturnsOnCall[i] = struct {
-		result1 error
-	}{result1}
-}
-
 func (fake *FakeRouter) SetNodeForRoom(arg1 string, arg2 string) error {
 	fake.setNodeForRoomMutex.Lock()
 	ret, specificReturn := fake.setNodeForRoomReturnsOnCall[len(fake.setNodeForRoomArgsForCall)]
@@ -752,7 +757,7 @@ func (fake *FakeRouter) StartReturnsOnCall(i int, result1 error) {
 	}{result1}
 }
 
-func (fake *FakeRouter) StartParticipantSignal(arg1 string, arg2 string, arg3 string, arg4 bool) (routing.MessageSink, routing.MessageSource, error) {
+func (fake *FakeRouter) StartParticipantSignal(arg1 string, arg2 string, arg3 string, arg4 bool) (string, routing.MessageSink, routing.MessageSource, error) {
 	fake.startParticipantSignalMutex.Lock()
 	ret, specificReturn := fake.startParticipantSignalReturnsOnCall[len(fake.startParticipantSignalArgsForCall)]
 	fake.startParticipantSignalArgsForCall = append(fake.startParticipantSignalArgsForCall, struct {
@@ -769,9 +774,9 @@ func (fake *FakeRouter) StartParticipantSignal(arg1 string, arg2 string, arg3 st
 		return stub(arg1, arg2, arg3, arg4)
 	}
 	if specificReturn {
-		return ret.result1, ret.result2, ret.result3
+		return ret.result1, ret.result2, ret.result3, ret.result4
 	}
-	return fakeReturns.result1, fakeReturns.result2, fakeReturns.result3
+	return fakeReturns.result1, fakeReturns.result2, fakeReturns.result3, fakeReturns.result4
 }
 
 func (fake *FakeRouter) StartParticipantSignalCallCount() int {
@@ -780,7 +785,7 @@ func (fake *FakeRouter) StartParticipantSignalCallCount() int {
 	return len(fake.startParticipantSignalArgsForCall)
 }
 
-func (fake *FakeRouter) StartParticipantSignalCalls(stub func(string, string, string, bool) (routing.MessageSink, routing.MessageSource, error)) {
+func (fake *FakeRouter) StartParticipantSignalCalls(stub func(string, string, string, bool) (string, routing.MessageSink, routing.MessageSource, error)) {
 	fake.startParticipantSignalMutex.Lock()
 	defer fake.startParticipantSignalMutex.Unlock()
 	fake.StartParticipantSignalStub = stub
@@ -793,33 +798,36 @@ func (fake *FakeRouter) StartParticipantSignalArgsForCall(i int) (string, string
 	return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3, argsForCall.arg4
 }
 
-func (fake *FakeRouter) StartParticipantSignalReturns(result1 routing.MessageSink, result2 routing.MessageSource, result3 error) {
+func (fake *FakeRouter) StartParticipantSignalReturns(result1 string, result2 routing.MessageSink, result3 routing.MessageSource, result4 error) {
 	fake.startParticipantSignalMutex.Lock()
 	defer fake.startParticipantSignalMutex.Unlock()
 	fake.StartParticipantSignalStub = nil
 	fake.startParticipantSignalReturns = struct {
-		result1 routing.MessageSink
-		result2 routing.MessageSource
-		result3 error
-	}{result1, result2, result3}
+		result1 string
+		result2 routing.MessageSink
+		result3 routing.MessageSource
+		result4 error
+	}{result1, result2, result3, result4}
 }
 
-func (fake *FakeRouter) StartParticipantSignalReturnsOnCall(i int, result1 routing.MessageSink, result2 routing.MessageSource, result3 error) {
+func (fake *FakeRouter) StartParticipantSignalReturnsOnCall(i int, result1 string, result2 routing.MessageSink, result3 routing.MessageSource, result4 error) {
 	fake.startParticipantSignalMutex.Lock()
 	defer fake.startParticipantSignalMutex.Unlock()
 	fake.StartParticipantSignalStub = nil
 	if fake.startParticipantSignalReturnsOnCall == nil {
 		fake.startParticipantSignalReturnsOnCall = make(map[int]struct {
-			result1 routing.MessageSink
-			result2 routing.MessageSource
-			result3 error
+			result1 string
+			result2 routing.MessageSink
+			result3 routing.MessageSource
+			result4 error
 		})
 	}
 	fake.startParticipantSignalReturnsOnCall[i] = struct {
-		result1 routing.MessageSink
-		result2 routing.MessageSource
-		result3 error
-	}{result1, result2, result3}
+		result1 string
+		result2 routing.MessageSink
+		result3 routing.MessageSource
+		result4 error
+	}{result1, result2, result3, result4}
 }
 
 func (fake *FakeRouter) Stop() {
@@ -904,6 +912,8 @@ func (fake *FakeRouter) Invocations() map[string][][]interface{} {
 	defer fake.invocationsMutex.RUnlock()
 	fake.clearRoomStateMutex.RLock()
 	defer fake.clearRoomStateMutex.RUnlock()
+	fake.createRTCSinkMutex.RLock()
+	defer fake.createRTCSinkMutex.RUnlock()
 	fake.getNodeMutex.RLock()
 	defer fake.getNodeMutex.RUnlock()
 	fake.getNodeForRoomMutex.RLock()
@@ -918,8 +928,6 @@ func (fake *FakeRouter) Invocations() map[string][][]interface{} {
 	defer fake.registerNodeMutex.RUnlock()
 	fake.removeDeadNodesMutex.RLock()
 	defer fake.removeDeadNodesMutex.RUnlock()
-	fake.sendRTCMessageMutex.RLock()
-	defer fake.sendRTCMessageMutex.RUnlock()
 	fake.setNodeForRoomMutex.RLock()
 	defer fake.setNodeForRoomMutex.RUnlock()
 	fake.startMutex.RLock()

pkg/rtc/participant.go

@@ -2,6 +2,7 @@ package rtc
 
 import (
 	"encoding/json"
+	"fmt"
 	"io"
 	"sync"
 	"sync/atomic"
@@ -225,7 +226,7 @@ func (p *ParticipantImpl) HandleOffer(sdp webrtc.SessionDescription) (answer web
 		"participant", p.Identity(),
 		//"sdp", sdp.SDP,
 	)
-	err = p.responseSink.WriteMessage(&livekit.SignalResponse{
+	p.writeMessage(&livekit.SignalResponse{
 		Message: &livekit.SignalResponse_Answer{
 			Answer: ToProtoSessionDescription(answer),
 		},
@@ -254,7 +255,7 @@ func (p *ParticipantImpl) AddTrack(clientId, name string, trackType livekit.Trac
 	}
 	p.pendingTracks[clientId] = ti
 
-	err := p.responseSink.WriteMessage(&livekit.SignalResponse{
+	p.writeMessage(&livekit.SignalResponse{
 		Message: &livekit.SignalResponse_TrackPublished{
 			TrackPublished: &livekit.TrackPublishedResponse{
 				Cid:   clientId,
@@ -262,10 +263,6 @@ func (p *ParticipantImpl) AddTrack(clientId, name string, trackType livekit.Trac
 			},
 		},
 	})
-	if err != nil {
-		logger.Errorw("could not write message", "error", err,
-			"participant", p.identity)
-	}
 }
 
 // handles a client answer response, with subscriber PC, server initiates the offer
@@ -324,9 +321,13 @@ func (p *ParticipantImpl) Close() error {
 	p.subscriber.pc.OnNegotiationNeeded(nil)
 	p.subscriber.pc.OnTrack(nil)
 	p.publisher.pc.OnICECandidate(nil)
+	// ensure this is synchronized
+	p.lock.RLock()
 	p.responseSink.Close()
-	if p.onClose != nil {
-		p.onClose(p)
+	onClose := p.onClose
+	p.lock.RUnlock()
+	if onClose != nil {
+		onClose(p)
 	}
 	p.publisher.Close()
 	p.subscriber.Close()
@@ -369,7 +370,7 @@ func (p *ParticipantImpl) RemoveSubscriber(participantId string) {
 // signal connection methods
 func (p *ParticipantImpl) SendJoinResponse(roomInfo *livekit.Room, otherParticipants []types.Participant) error {
 	// send Join response
-	return p.responseSink.WriteMessage(&livekit.SignalResponse{
+	return p.writeMessage(&livekit.SignalResponse{
 		Message: &livekit.SignalResponse_Join{
 			Join: &livekit.JoinResponse{
 				Room:              roomInfo,
@@ -386,7 +387,7 @@ func (p *ParticipantImpl) SendParticipantUpdate(participants []*livekit.Particip
 		return nil
 	}
 
-	return p.responseSink.WriteMessage(&livekit.SignalResponse{
+	return p.writeMessage(&livekit.SignalResponse{
 		Message: &livekit.SignalResponse_Update{
 			Update: &livekit.ParticipantUpdate{
 				Participants: participants,
@@ -400,7 +401,7 @@ func (p *ParticipantImpl) SendActiveSpeakers(speakers []*livekit.SpeakerInfo) er
 		return nil
 	}
 
-	return p.responseSink.WriteMessage(&livekit.SignalResponse{
+	return p.writeMessage(&livekit.SignalResponse{
 		Message: &livekit.SignalResponse_Speaker{
 			Speaker: &livekit.ActiveSpeakerUpdate{
 				Speakers: speakers,
@@ -485,20 +486,17 @@ func (p *ParticipantImpl) sendIceCandidate(c *webrtc.ICECandidate, target liveki
 	//logger.Debugw("sending ice candidates")
 	trickle := ToProtoTrickle(ci)
 	trickle.Target = target
-	err := p.responseSink.WriteMessage(&livekit.SignalResponse{
+	p.writeMessage(&livekit.SignalResponse{
 		Message: &livekit.SignalResponse_Trickle{
 			Trickle: trickle,
 		},
 	})
-	if err != nil {
-		logger.Errorw("could not send trickle", "err", err,
-			"participant", p.identity)
-	}
 }
 
 // initiates server-driven negotiation by creating an offer
 func (p *ParticipantImpl) negotiate() {
 	if p.State() == livekit.ParticipantInfo_DISCONNECTED {
+		logger.Debugw("skipping server negotiation", "participant", p.Identity())
 		// skip when disconnected
 		return
 	}
@@ -521,16 +519,12 @@ func (p *ParticipantImpl) negotiate() {
 		"participant", p.Identity(),
 		//"sdp", offer.SDP,
 	)
-	err = p.responseSink.WriteMessage(&livekit.SignalResponse{
+
+	p.writeMessage(&livekit.SignalResponse{
 		Message: &livekit.SignalResponse_Offer{
 			Offer: ToProtoSessionDescription(offer),
 		},
 	})
-	if err != nil {
-		logger.Errorw("could not send offer to participant",
-			"err", err,
-			"participant", p.identity)
-	}
 }
 
 func (p *ParticipantImpl) updateState(state livekit.ParticipantInfo_State) {
@@ -548,6 +542,23 @@ func (p *ParticipantImpl) updateState(state livekit.ParticipantInfo_State) {
 	}
 }
 
+func (p *ParticipantImpl) writeMessage(msg *livekit.SignalResponse) error {
+	if p.State() == livekit.ParticipantInfo_DISCONNECTED {
+		return nil
+	}
+	sink := p.responseSink
+	err := sink.WriteMessage(msg)
+	if err != nil {
+		logger.Warnw("could not send message to participant",
+			"error", err,
+			"id", p.ID(),
+			"participant", p.identity,
+			"message", fmt.Sprintf("%T", msg.Message))
+		return err
+	}
+	return nil
+}
+
 // when a new remoteTrack is created, creates a Track and adds it to room
 func (p *ParticipantImpl) onMediaTrack(track *webrtc.TrackRemote, rtpReceiver *webrtc.RTPReceiver) {
 	logger.Debugw("mediaTrack added", "participant", p.Identity(), "remoteTrack", track.ID())

pkg/rtc/transport.go

@@ -8,6 +8,7 @@ import (
 	"github.com/bep/debounce"
 	"github.com/pion/webrtc/v3"
 
+	"github.com/livekit/livekit-server/pkg/logger"
 	"github.com/livekit/livekit-server/proto/livekit"
 )
 
@@ -111,6 +112,7 @@ func (t *PCTransport) SetRemoteDescription(sd webrtc.SessionDescription) error {
 	state := t.negotiationState.Load().(int)
 	t.negotiationState.Store(negotiationStateNone)
 	if state == negotiationStateServer && t.onNegotiation != nil {
+		logger.Debugw("negotiating again")
 		// need to negotiate again
 		t.negotiate()
 	}
@@ -126,14 +128,14 @@ func (t *PCTransport) negotiate() {
 	t.debouncedNegotiate(func() {
 		state := t.negotiationState.Load().(int)
 		// when there's an ongoing negotiation, let it finish and not disrupt its state
-		if state != negotiationStateNone {
+		if state == negotiationStateClient {
+			logger.Debugw("skipping negotiation, trying again later")
 			t.negotiationState.Store(negotiationStateServer)
 			return
 		}
 
 		if t.onNegotiation != nil {
 			t.onNegotiation()
-
 			// indicate waiting for client
 			t.negotiationState.Store(negotiationStateClient)
 		}

pkg/service/roomservice.go

@@ -106,7 +106,12 @@ func (s *RoomService) RemoveParticipant(ctx context.Context, req *livekit.RoomPa
 		return
 	}
 
-	err = s.roomManager.router.SendRTCMessage(req.Room, participant.Identity, &livekit.RTCNodeMessage{
+	rtcSink, err := s.roomManager.router.CreateRTCSink(req.Room, participant.Identity)
+	if err != nil {
+		return
+	}
+	defer rtcSink.Close()
+	err = rtcSink.WriteMessage(&livekit.RTCNodeMessage{
 		Message: &livekit.RTCNodeMessage_RemoveParticipant{
 			RemoveParticipant: req,
 		},
@@ -126,7 +131,12 @@ func (s *RoomService) MutePublishedTrack(ctx context.Context, req *livekit.MuteR
 		return
 	}
 
-	err = s.roomManager.router.SendRTCMessage(req.Room, participant.Identity, &livekit.RTCNodeMessage{
+	rtcSink, err := s.roomManager.router.CreateRTCSink(req.Room, participant.Identity)
+	if err != nil {
+		return
+	}
+	defer rtcSink.Close()
+	err = rtcSink.WriteMessage(&livekit.RTCNodeMessage{
 		Message: &livekit.RTCNodeMessage_MuteTrack{
 			MuteTrack: req,
 		},

pkg/service/rtcservice.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"io"
 	"net/http"
+	"strings"
 
 	"github.com/gorilla/websocket"
 
@@ -84,15 +85,16 @@ func (s *RTCService) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	}
 
 	// this needs to be started first *before* using router functions on this node
-	reqSink, resSource, err := s.router.StartParticipantSignal(roomName, identity, metadata, isReconnect)
+	connId, reqSink, resSource, err := s.router.StartParticipantSignal(roomName, identity, metadata, isReconnect)
 	if err != nil {
 		handleError(w, http.StatusInternalServerError, "could not start session: "+err.Error())
 		return
 	}
 
 	done := make(chan bool, 1)
+	// function exits when websocket terminates, it'll close the event reading off of response sink as well
 	defer func() {
-		logger.Infow("WS connection closed", "participant", identity)
+		logger.Infow("WS connection closed", "participant", identity, "connectionId", connId)
 		reqSink.Close()
 		close(done)
 	}()
@@ -109,6 +111,7 @@ func (s *RTCService) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	sigConn := NewWSSignalConnection(conn)
 
 	logger.Infow("new client WS connected",
+		"connectionId", connId,
 		"room", rm.Sid,
 		"roomName", rm.Name,
 		"name", identity,
@@ -128,14 +131,16 @@ func (s *RTCService) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 				return
 			case msg := <-resSource.ReadChan():
 				if msg == nil {
-					logger.Errorw("source closed connection", "participant", identity)
+					logger.Infow("source closed connection", "participant", identity,
+						"connectionId", connId)
 					return
 				}
 				res, ok := msg.(*livekit.SignalResponse)
 				if !ok {
 					logger.Errorw("unexpected message type",
 						"type", fmt.Sprintf("%T", msg),
-						"participant", identity)
+						"participant", identity,
+						"connectionId", connId)
 					continue
 				}
 
@@ -151,15 +156,17 @@ func (s *RTCService) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	for {
 		req, err := sigConn.ReadRequest()
 		// normal closure
-		if err == io.EOF || websocket.IsCloseError(err, websocket.CloseAbnormalClosure, websocket.CloseGoingAway, websocket.CloseNormalClosure) {
-			return
-		} else if err != nil {
-			logger.Errorw("error reading from websocket", "error", err)
-			return
+		if err != nil {
+			if err == io.EOF || strings.HasSuffix(err.Error(), "use of closed network connection") || websocket.IsCloseError(err, websocket.CloseAbnormalClosure, websocket.CloseGoingAway, websocket.CloseNormalClosure) {
+				return
+			} else {
+				logger.Errorw("error reading from websocket", "error", err)
+				return
+			}
 		}
-
-		if err = reqSink.WriteMessage(req); err != nil {
-			logger.Warnw("error writing to request sink", "error", err)
+		if err := reqSink.WriteMessage(req); err != nil {
+			logger.Warnw("error writing to request sink", "error", err,
+				"participant", identity, "connectionId", connId)
 		}
 	}
 }

test/multinode_test.go

@@ -51,7 +51,7 @@ func TestMultiNodeRouting(t *testing.T) {
 		}
 
 		tr1 := c2.SubscribedTracks()[c1.ID()][0]
-		assert.Equal(t, "webcam", tr1.StreamID())
+		assert.Equal(t, c1.ID(), tr1.StreamID())
 		return true
 	})
 

test/singlenode_test.go

@@ -68,7 +68,7 @@ func TestSinglePublisher(t *testing.T) {
 		}
 
 		tr1 := c2.SubscribedTracks()[c1.ID()][0]
-		assert.Equal(t, "webcam", tr1.StreamID())
+		assert.Equal(t, c1.ID(), tr1.StreamID())
 		return true
 	})
 	if !success {
@@ -94,6 +94,6 @@ func TestSinglePublisher(t *testing.T) {
 	// ensure that the track ids are generated by server
 	tracks := c3.SubscribedTracks()[c1.ID()]
 	for _, tr := range tracks {
-		assert.True(t, strings.Contains(tr.ID(), "|TR_"), "track should begin with TR")
+		assert.True(t, strings.HasPrefix(tr.ID(), "TR_"), "track should begin with TR")
 	}
 }