fixed multi-node so signal node can start sessions

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", "./...", "-count=1")
+	cmd := exec.Command("go", "test", "./...", "-count=1", "-timeout=1m")
 	connectStd(cmd)
 	return cmd.Run()
 }

pkg/routing/errors.go

@@ -6,7 +6,6 @@ var (
 	ErrNotFound             = errors.New("could not find object")
 	ErrHandlerNotDefined    = errors.New("handler not defined")
 	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/redisrouter.go

@@ -161,12 +161,7 @@ func (r *RedisRouter) StartParticipant(roomName, participantId, participantName
 		return err
 	}
 
-	// StartParticipant should only be called on the current node
-	if rtcNode != r.currentNode.Id {
-		return ErrIncorrectNodeForRoom
-	}
-
-	if r.useLocal {
+	if r.useLocal && rtcNode == r.currentNode.Id {
 		return r.LocalRouter.StartParticipant(roomName, participantId, participantId)
 	}
 
@@ -282,7 +277,6 @@ func (r *RedisRouter) subscribeWorker() {
 			continue
 		}
 
-		logger.Debugw("received redis message", "message", msg, "node", r.currentNode.Id)
 		rm := livekit.RouterMessage{}
 		err = proto.Unmarshal([]byte(msg.Payload), &rm)
 		pId := rm.ParticipantId

pkg/rtc/participant.go

@@ -109,7 +109,7 @@ func NewParticipant(participantId, name string, pc types.PeerConnection, rs rout
 		ci := c.ToJSON()
 
 		// write candidate
-		logger.Debugw("sending ice candidates")
+		//logger.Debugw("sending ice candidates")
 		err := rs.WriteMessage(&livekit.SignalResponse{
 			Message: &livekit.SignalResponse_Trickle{
 				Trickle: ToProtoTrickle(ci),

pkg/service/roommanager.go

@@ -54,7 +54,8 @@ func (r *RoomManager) CreateRoom(req *livekit.CreateRoomRequest) (*livekit.Room,
 		return nil, err
 	}
 
-	if req.NodeId == "" {
+	nodeId := req.NodeId
+	if nodeId == "" {
 		// select a node for room
 		nodes, err := r.router.ListNodes()
 		if err != nil {
@@ -65,10 +66,11 @@ func (r *RoomManager) CreateRoom(req *livekit.CreateRoomRequest) (*livekit.Room,
 		if err != nil {
 			return nil, err
 		}
-		req.NodeId = node.Id
+		nodeId = node.Id
 	}
 
-	if err := r.router.SetNodeForRoom(req.Name, req.NodeId); err != nil {
+	logger.Debugw("selected node for room", "room", rm.Name, "node", nodeId)
+	if err := r.router.SetNodeForRoom(req.Name, nodeId); err != nil {
 		return nil, err
 	}
 
@@ -131,6 +133,7 @@ func (r *RoomManager) StartSession(roomName, participantId, participantName stri
 
 	logger.Debugw("starting RTC session",
 		"room", roomName,
+		"node", r.currentNode.Id,
 		"participant", participantName,
 		"num_participants", len(room.GetParticipants()),
 	)
@@ -201,7 +204,6 @@ func (r *RoomManager) rtcSessionWorker(room *rtc.Room, participant types.Partici
 	}()
 	defer rtc.Recover()
 
-	logger.Debugw("starting RTC session", "node", r.currentNode.Id)
 	for {
 		select {
 		case <-time.After(time.Millisecond * 100):

pkg/service/rtcservice.go

@@ -70,6 +70,30 @@ func (s *RTCService) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
+	participantId := utils.NewGuid(utils.ParticipantPrefix)
+	err = s.router.StartParticipant(roomName, participantId, pName)
+	if err != nil {
+		handleError(w, http.StatusInternalServerError, "could not start session: "+err.Error())
+		return
+	}
+	reqSink, err := s.router.GetRequestSink(participantId)
+	if err != nil {
+		handleError(w, http.StatusInternalServerError, "could not get request sink"+err.Error())
+		return
+	}
+	done := make(chan bool, 1)
+	defer func() {
+		logger.Infow("WS connection closed", "participant", pName)
+		reqSink.Close()
+		close(done)
+	}()
+
+	resSource, err := s.router.GetResponseSource(participantId)
+	if err != nil {
+		handleError(w, http.StatusInternalServerError, "could not get response source"+err.Error())
+		return
+	}
+
 	// upgrade only once the basics are good to go
 	conn, err := s.upgrader.Upgrade(w, r, nil)
 	if err != nil {
@@ -79,39 +103,14 @@ func (s *RTCService) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		handleError(w, http.StatusInternalServerError, err.Error())
 		return
 	}
-
 	sigConn := NewWSSignalConnection(conn)
 
-	participantId := utils.NewGuid(utils.ParticipantPrefix)
-	err = s.router.StartParticipant(roomName, participantId, pName)
-	if err != nil {
-		handleError(w, http.StatusInternalServerError, "could not start session: "+err.Error())
-		return
-	}
-
 	logger.Infow("new client connected",
 		"room", rm.Sid,
 		"roomName", rm.Name,
 		"name", pName,
 	)
 
-	reqSink, err := s.router.GetRequestSink(participantId)
-	if err != nil {
-		handleError(w, http.StatusInternalServerError, "could not get request sink"+err.Error())
-		return
-	}
-	resSource, err := s.router.GetResponseSource(participantId)
-	if err != nil {
-		handleError(w, http.StatusInternalServerError, "could not get response source"+err.Error())
-		return
-	}
-	done := make(chan bool, 1)
-	defer func() {
-		logger.Infow("WS connection closed", "participant", pName)
-		reqSink.Close()
-		close(done)
-	}()
-
 	// handle responses
 	go func() {
 		for {

test/integration_helpers.go

@@ -115,10 +115,10 @@ func waitUntilConnected(t *testing.T, clients ...*client.RTCClient) {
 		c := clients[i]
 		wg.Add(1)
 		go func() {
+			defer wg.Done()
 			if !assert.NoError(t, c.WaitUntilConnected()) {
 				t.Fatal("one or more clients could not connect")
 			}
-			wg.Done()
 		}()
 	}
 	wg.Wait()

test/multinode_test.go

@@ -31,13 +31,15 @@ func TestMultiNodeRouting(t *testing.T) {
 
 	// one node connecting to node 1, and another connecting to node 2
 	c1 := createRTCClient("c1", defaultServerPort)
-	c2 := createRTCClient("c2", defaultServerPort)
+	c2 := createRTCClient("c2", secondServerPort)
 	waitUntilConnected(t, c1, c2)
 
 	// c1 publishing, and c2 receiving
 	t1, err := c1.AddStaticTrack("audio/opus", "audio", "webcam")
 	assert.NoError(t, err)
-	defer t1.Stop()
+	if t1 != nil {
+		defer t1.Stop()
+	}
 
 	withTimeout(t, "c2 should receive one track", func() bool {
 		if len(c2.SubscribedTracks()) == 0 {

test/scenarios.go

@@ -0,0 +1,10 @@
+package test
+
+import (
+	"testing"
+)
+
+// a scenario with lots of clients connecting, publishing, and leaving at random periods
+func scenarioRandom(t *testing.T) {
+
+}