fixed panic when client connected abruptly

cmd/cli/client/client.go

@@ -335,6 +335,8 @@ func (c *RTCClient) ResumeLogs() {
 }
 
 func (c *RTCClient) SendRequest(msg *livekit.SignalRequest) error {
+	c.lock.Lock()
+	defer c.lock.Unlock()
 	payload, err := protojson.Marshal(msg)
 	if err != nil {
 		return err

magefile.go

@@ -157,7 +157,7 @@ func Test() error {
 // run all thests including integration
 func TestAll() error {
 	mg.Deps(Proto)
-	cmd := exec.Command("go", "test", "./...")
+	cmd := exec.Command("go", "test", "./...", "-count=1")
 	connectStd(cmd)
 	return cmd.Run()
 }

pkg/routing/errors.go

@@ -8,4 +8,5 @@ var (
 	ErrNoAvailableNodes     = errors.New("could not find any available nodes")
 	ErrIncorrectNodeForRoom = errors.New("incorrect node for the current room")
 	errInvalidRouterMessage = errors.New("invalid router message")
+	ErrChannelClosed        = errors.New("channel closed")
 )

pkg/routing/messagechannel.go

@@ -5,13 +5,15 @@ import (
 )
 
 type MessageChannel struct {
-	msgChan chan proto.Message
-	onClose func()
+	msgChan  chan proto.Message
+	isClosed bool
+	onClose  func()
 }
 
 func NewMessageChannel() *MessageChannel {
 	return &MessageChannel{
-		msgChan: make(chan proto.Message, 1),
+		// allow some buffer to avoid blocked writes
+		msgChan: make(chan proto.Message, 2),
 	}
 }
 
@@ -20,6 +22,9 @@ func (m *MessageChannel) OnClose(f func()) {
 }
 
 func (m *MessageChannel) WriteMessage(msg proto.Message) error {
+	if m.isClosed {
+		return ErrChannelClosed
+	}
 	m.msgChan <- msg
 	return nil
 }
@@ -29,6 +34,7 @@ func (m *MessageChannel) ReadChan() <-chan proto.Message {
 }
 
 func (m *MessageChannel) Close() {
+	m.isClosed = true
 	close(m.msgChan)
 	if m.onClose != nil {
 		m.onClose()

pkg/routing/redis.go

@@ -69,6 +69,7 @@ type RedisSink struct {
 	rc            *redis.Client
 	nodeId        string
 	participantId string
+	isClosed      bool
 	once          sync.Once
 	onClose       func()
 }
@@ -83,12 +84,16 @@ func NewRedisSink(rc *redis.Client, nodeId, participantId string) *RedisSink {
 }
 
 func (s *RedisSink) WriteMessage(msg proto.Message) error {
+	if s.isClosed {
+		return ErrChannelClosed
+	}
 	return publishRouterMessage(s.rc, s.nodeId, s.participantId, msg)
 }
 
 func (s *RedisSink) Close() {
 	s.once.Do(func() {
 		publishRouterMessage(s.rc, s.nodeId, s.participantId, &livekit.EndSession{})
+		s.isClosed = true
 		if s.onClose != nil {
 			s.onClose()
 		}

pkg/rtc/participant.go

@@ -320,7 +320,6 @@ func (p *ParticipantImpl) Close() error {
 	if p.ctx.Err() != nil {
 		return p.ctx.Err()
 	}
-	close(p.rtcpCh)
 	p.onICECandidate = nil
 	p.peerConn.OnDataChannel(nil)
 	p.peerConn.OnICECandidate(nil)
@@ -332,6 +331,7 @@ func (p *ParticipantImpl) Close() error {
 		p.onClose(p)
 	}
 	p.cancel()
+	close(p.rtcpCh)
 	return p.peerConn.Close()
 }
 
@@ -434,6 +434,10 @@ func (p *ParticipantImpl) scheduleNegotiate() {
 
 // initiates server-driven negotiation by creating an offer
 func (p *ParticipantImpl) negotiate() {
+	if p.state == livekit.ParticipantInfo_DISCONNECTED {
+		// skip when disconnected
+		return
+	}
 	p.negotiationCond.L.Lock()
 	for p.negotiationState != negotiationStateNone {
 		p.negotiationCond.Wait()

pkg/rtc/participant_internal_test.go

@@ -5,6 +5,7 @@ import (
 
 	"github.com/stretchr/testify/assert"
 
+	"github.com/livekit/livekit-server/pkg/routing"
 	"github.com/livekit/livekit-server/pkg/routing/routingfakes"
 	"github.com/livekit/livekit-server/pkg/rtc/types"
 	"github.com/livekit/livekit-server/pkg/rtc/types/typesfakes"
@@ -68,6 +69,26 @@ func TestTrackPublishEvents(t *testing.T) {
 	assert.True(t, updated)
 }
 
+// after disconnection, things should continue to function and not panic
+func TestDisconnectTiming(t *testing.T) {
+	t.Run("negotiate doesn't fail after channel closed", func(t *testing.T) {
+		p := newParticipantForTest("test")
+		msg := routing.NewMessageChannel()
+		p.responseSink = msg
+		go func() {
+			for msg := range msg.ReadChan() {
+				t.Log("received message from chan", msg)
+			}
+		}()
+		track := &typesfakes.FakePublishedTrack{}
+		p.handleTrackPublished(track)
+
+		// close channel and then try to negotiate
+		msg.Close()
+		p.negotiate()
+	})
+}
+
 func newParticipantForTest(name string) *ParticipantImpl {
 	p, _ := NewParticipant(
 		utils.NewGuid(utils.ParticipantPrefix),

pkg/rtc/room.go

@@ -194,8 +194,8 @@ func (r *Room) broadcastParticipantState(p types.Participant) {
 
 	updates := ToProtoParticipants([]types.Participant{p})
 	for _, op := range r.participants {
-		// skip itself
-		if p.ID() == op.ID() {
+		// skip itself && closed participants
+		if p.ID() == op.ID() || p.State() == livekit.ParticipantInfo_DISCONNECTED {
 			continue
 		}
 

pkg/rtc/room_test.go

@@ -56,22 +56,27 @@ func TestRoomJoin(t *testing.T) {
 	})
 
 	t.Run("participant state change is broadcasted to others", func(t *testing.T) {
-		rm := newRoomWithParticipants(t, 1)
+		rm := newRoomWithParticipants(t, numParticipants)
 		participants := rm.GetParticipants()
 		p := participants[0].(*typesfakes.FakeParticipant)
+		disconnectedParticipant := participants[1].(*typesfakes.FakeParticipant)
+		disconnectedParticipant.StateReturns(livekit.ParticipantInfo_DISCONNECTED)
 
 		rm.RemoveParticipant(p.ID())
 		p.OnStateChangeArgsForCall(0)(p, livekit.ParticipantInfo_ACTIVE)
 		time.Sleep(defaultDelay)
 
+		numUpdates := 0
 		for _, op := range participants {
-			if op == p {
+			if op == p || op == disconnectedParticipant {
 				assert.Zero(t, p.SendParticipantUpdateCallCount())
 				continue
 			}
 			fakeP := op.(*typesfakes.FakeParticipant)
 			assert.Equal(t, 1, fakeP.SendParticipantUpdateCallCount())
+			numUpdates += 1
 		}
+		assert.Equal(t, numParticipants-2, numUpdates)
 	})
 
 	t.Run("cannot exceed max participants", func(t *testing.T) {

pkg/service/roommanager.go

@@ -121,7 +121,7 @@ func (r *RoomManager) Cleanup() error {
 	return nil
 }
 
-// starts WebRTC session when a new participant is connected
+// starts WebRTC session when a new participant is connected, takes place on RTC node
 func (r *RoomManager) StartSession(roomName, participantId, participantName string, requestSource routing.MessageSource, responseSink routing.MessageSink) {
 	room, err := r.getOrCreateRoom(roomName)
 	if err != nil {