add roommanager service (#2931)

* add roommanager service

* cleanup

* deps

* timeout

* cleanup

* cleanup

* config

* cleanup

go.mod

@@ -19,7 +19,7 @@ require (
 	github.com/jxskiss/base62 v1.1.0
 	github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1
 	github.com/livekit/mediatransportutil v0.0.0-20240730083616-559fa5ece598
-	github.com/livekit/protocol v1.19.4-0.20240813090210-bbd53d6fbfdd
+	github.com/livekit/protocol v1.20.1-0.20240813123848-0072ee0c6e47
 	github.com/livekit/psrpc v0.5.3-0.20240616012458-ac39c8549a0a
 	github.com/mackerelio/go-osstat v0.2.5
 	github.com/magefile/mage v1.15.0

go.sum

@@ -167,8 +167,8 @@ github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1 h1:jm09419p0lqTkD
 github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1/go.mod h1:Rs3MhFwutWhGwmY1VQsygw28z5bWcnEYmS1OG9OxjOQ=
 github.com/livekit/mediatransportutil v0.0.0-20240730083616-559fa5ece598 h1:yLlkHk2feSLHstD9n4VKg7YEBR4rLODTI4WE8gNBEnQ=
 github.com/livekit/mediatransportutil v0.0.0-20240730083616-559fa5ece598/go.mod h1:jwKUCmObuiEDH0iiuJHaGMXwRs3RjrB4G6qqgkr/5oE=
-github.com/livekit/protocol v1.19.4-0.20240813090210-bbd53d6fbfdd h1:ejCQXUVF/sj16cRIwR0cN/zf0OBeWfQ9OAy6yEUYohc=
-github.com/livekit/protocol v1.19.4-0.20240813090210-bbd53d6fbfdd/go.mod h1:AFuwk3+uIWFeO5ohKjx5w606Djl940+wktaZ441VoCI=
+github.com/livekit/protocol v1.20.1-0.20240813123848-0072ee0c6e47 h1:Hsur+//Q0Ll/JfWydKpXxjwDF0FnTzJYOsL0D7s8sAc=
+github.com/livekit/protocol v1.20.1-0.20240813123848-0072ee0c6e47/go.mod h1:AFuwk3+uIWFeO5ohKjx5w606Djl940+wktaZ441VoCI=
 github.com/livekit/psrpc v0.5.3-0.20240616012458-ac39c8549a0a h1:EQAHmcYEGlc6V517cQ3Iy0+jHgP6+tM/B4l2vGuLpQo=
 github.com/livekit/psrpc v0.5.3-0.20240616012458-ac39c8549a0a/go.mod h1:CQUBSPfYYAaevg1TNCc6/aYsa8DJH4jSRFdCeSZk5u0=
 github.com/mackerelio/go-osstat v0.2.5 h1:+MqTbZUhoIt4m8qzkVoXUJg1EuifwlAJSk4Yl2GXh+o=

pkg/config/config.go

@@ -242,6 +242,9 @@ type RoomConfig struct {
 	EnableRemoteUnmute bool               `yaml:"enable_remote_unmute,omitempty"`
 	PlayoutDelay       PlayoutDelayConfig `yaml:"playout_delay,omitempty"`
 	SyncStreams        bool               `yaml:"sync_streams,omitempty"`
+	CreateRoomEnabled  bool               `yaml:"create_room_enabled,omitempty"`
+	CreateRoomTimeout  time.Duration      `yaml:"create_room_timeout,omitempty"`
+	CreateRoomAttempts int                `yaml:"create_room_attempts,omitempty"`
 	// deprecated, moved to limits
 	MaxMetadataSize uint32 `yaml:"max_metadata_size,omitempty"`
 	// deprecated, moved to limits
@@ -537,8 +540,10 @@ var DefaultConfig = Config{
 			{Mime: webrtc.MimeTypeVP9},
 			{Mime: webrtc.MimeTypeAV1},
 		},
-		EmptyTimeout:     5 * 60,
-		DepartureTimeout: 20,
+		EmptyTimeout:       5 * 60,
+		DepartureTimeout:   20,
+		CreateRoomTimeout:  10 * time.Second,
+		CreateRoomAttempts: 3,
 	},
 	Limit: LimitConfig{
 		MaxMetadataSize:              64000,

pkg/routing/interfaces.go

@@ -62,6 +62,7 @@ type ParticipantInit struct {
 	ID                   livekit.ParticipantID
 	SubscriberAllowPause *bool
 	DisableICELite       bool
+	CreateRoom           *livekit.CreateRoomRequest
 }
 
 // Router allows multiple nodes to coordinate the participant session
@@ -96,12 +97,20 @@ type StartParticipantSignalResults struct {
 }
 
 type MessageRouter interface {
+	// CreateRoom starts an rtc room
+	CreateRoom(ctx context.Context, req *livekit.CreateRoomRequest) (res *livekit.Room, err error)
 	// StartParticipantSignal participant signal connection is ready to start
 	StartParticipantSignal(ctx context.Context, roomName livekit.RoomName, pi ParticipantInit) (res StartParticipantSignalResults, err error)
 }
 
-func CreateRouter(rc redis.UniversalClient, node LocalNode, signalClient SignalClient, kps rpc.KeepalivePubSub) Router {
-	lr := NewLocalRouter(node, signalClient)
+func CreateRouter(
+	rc redis.UniversalClient,
+	node LocalNode,
+	signalClient SignalClient,
+	roomManagerClient RoomManagerClient,
+	kps rpc.KeepalivePubSub,
+) Router {
+	lr := NewLocalRouter(node, signalClient, roomManagerClient)
 
 	if rc != nil {
 		return NewRedisRouter(lr, rc, kps)
@@ -132,6 +141,7 @@ func (pi *ParticipantInit) ToStartSession(roomName livekit.RoomName, connectionI
 		AdaptiveStream:  pi.AdaptiveStream,
 		ParticipantId:   string(pi.ID),
 		DisableIceLite:  pi.DisableICELite,
+		CreateRoom:      pi.CreateRoom,
 	}
 	if pi.SubscriberAllowPause != nil {
 		subscriberAllowPause := *pi.SubscriberAllowPause

pkg/routing/localrouter.go

@@ -30,8 +30,9 @@ var _ Router = (*LocalRouter)(nil)
 
 // a router of messages on the same node, basic implementation for local testing
 type LocalRouter struct {
-	currentNode  LocalNode
-	signalClient SignalClient
+	currentNode       LocalNode
+	signalClient      SignalClient
+	roomManagerClient RoomManagerClient
 
 	lock sync.RWMutex
 	// channels for each participant
@@ -40,12 +41,17 @@ type LocalRouter struct {
 	isStarted        atomic.Bool
 }
 
-func NewLocalRouter(currentNode LocalNode, signalClient SignalClient) *LocalRouter {
+func NewLocalRouter(
+	currentNode LocalNode,
+	signalClient SignalClient,
+	roomManagerClient RoomManagerClient,
+) *LocalRouter {
 	return &LocalRouter{
-		currentNode:      currentNode,
-		signalClient:     signalClient,
-		requestChannels:  make(map[string]*MessageChannel),
-		responseChannels: make(map[string]*MessageChannel),
+		currentNode:       currentNode,
+		signalClient:      signalClient,
+		roomManagerClient: roomManagerClient,
+		requestChannels:   make(map[string]*MessageChannel),
+		responseChannels:  make(map[string]*MessageChannel),
 	}
 }
 
@@ -89,6 +95,14 @@ func (r *LocalRouter) ListNodes() ([]*livekit.Node, error) {
 	}, nil
 }
 
+func (r *LocalRouter) CreateRoom(ctx context.Context, req *livekit.CreateRoomRequest) (res *livekit.Room, err error) {
+	return r.CreateRoomWithNodeID(ctx, req, livekit.NodeID(r.currentNode.Id))
+}
+
+func (r *LocalRouter) CreateRoomWithNodeID(ctx context.Context, req *livekit.CreateRoomRequest, nodeID livekit.NodeID) (res *livekit.Room, err error) {
+	return r.roomManagerClient.CreateRoom(ctx, nodeID, req)
+}
+
 func (r *LocalRouter) StartParticipantSignal(ctx context.Context, roomName livekit.RoomName, pi ParticipantInit) (res StartParticipantSignalResults, err error) {
 	return r.StartParticipantSignalWithNodeID(ctx, roomName, pi, livekit.NodeID(r.currentNode.Id))
 }

pkg/routing/redisrouter.go

@@ -109,6 +109,7 @@ func (r *RedisRouter) RemoveDeadNodes() error {
 	return nil
 }
 
+// GetNodeForRoom finds the node where the room is hosted at
 func (r *RedisRouter) GetNodeForRoom(_ context.Context, roomName livekit.RoomName) (*livekit.Node, error) {
 	nodeID, err := r.rc.HGet(r.ctx, NodeRoomKey, string(roomName)).Result()
 	if err == redis.Nil {
@@ -161,9 +162,17 @@ func (r *RedisRouter) ListNodes() ([]*livekit.Node, error) {
 	return nodes, nil
 }
 
+func (r *RedisRouter) CreateRoom(ctx context.Context, req *livekit.CreateRoomRequest) (res *livekit.Room, err error) {
+	rtcNode, err := r.GetNodeForRoom(ctx, livekit.RoomName(req.Name))
+	if err != nil {
+		return
+	}
+
+	return r.CreateRoomWithNodeID(ctx, req, livekit.NodeID(rtcNode.Id))
+}
+
 // StartParticipantSignal signal connection sets up paths to the RTC node, and starts to route messages to that message queue
 func (r *RedisRouter) StartParticipantSignal(ctx context.Context, roomName livekit.RoomName, pi ParticipantInit) (res StartParticipantSignalResults, err error) {
-	// find the node where the room is hosted at
 	rtcNode, err := r.GetNodeForRoom(ctx, roomName)
 	if err != nil {
 		return

pkg/routing/roommanager.go

@@ -0,0 +1,61 @@
+// Copyright 2023 LiveKit, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package routing
+
+import (
+	"context"
+
+	"github.com/livekit/livekit-server/pkg/config"
+	"github.com/livekit/protocol/livekit"
+	"github.com/livekit/protocol/rpc"
+	"github.com/livekit/psrpc"
+	"github.com/livekit/psrpc/pkg/middleware"
+)
+
+//go:generate go run github.com/maxbrunsfeld/counterfeiter/v6 -generate
+
+//counterfeiter:generate . RoomManagerClient
+type RoomManagerClient interface {
+	rpc.TypedRoomManagerClient
+}
+
+type roomManagerClient struct {
+	config config.RoomConfig
+	client rpc.TypedRoomManagerClient
+}
+
+func NewRoomManagerClient(clientParams rpc.ClientParams, config config.RoomConfig) (RoomManagerClient, error) {
+	c, err := rpc.NewTypedRoomManagerClient(
+		clientParams.Bus,
+		psrpc.WithClientChannelSize(clientParams.BufferSize),
+		middleware.WithClientMetrics(clientParams.Observer),
+		rpc.WithClientLogger(clientParams.Logger),
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	return &roomManagerClient{
+		config: config,
+		client: c,
+	}, nil
+}
+
+func (c *roomManagerClient) CreateRoom(ctx context.Context, nodeID livekit.NodeID, req *livekit.CreateRoomRequest, opts ...psrpc.RequestOption) (*livekit.Room, error) {
+	return c.client.CreateRoom(ctx, nodeID, req, append(opts, psrpc.WithRequestInterceptors(middleware.NewRPCRetryInterceptor(middleware.RetryOptions{
+		MaxAttempts: c.config.CreateRoomAttempts,
+		Timeout:     c.config.CreateRoomTimeout,
+	})))...)
+}

pkg/routing/routingfakes/fake_room_manager_client.go

@@ -0,0 +1,125 @@
+// Code generated by counterfeiter. DO NOT EDIT.
+package routingfakes
+
+import (
+	"context"
+	"sync"
+
+	"github.com/livekit/livekit-server/pkg/routing"
+	"github.com/livekit/protocol/livekit"
+	"github.com/livekit/psrpc"
+)
+
+type FakeRoomManagerClient struct {
+	CreateRoomStub        func(context.Context, livekit.NodeID, *livekit.CreateRoomRequest, ...psrpc.RequestOption) (*livekit.Room, error)
+	createRoomMutex       sync.RWMutex
+	createRoomArgsForCall []struct {
+		arg1 context.Context
+		arg2 livekit.NodeID
+		arg3 *livekit.CreateRoomRequest
+		arg4 []psrpc.RequestOption
+	}
+	createRoomReturns struct {
+		result1 *livekit.Room
+		result2 error
+	}
+	createRoomReturnsOnCall map[int]struct {
+		result1 *livekit.Room
+		result2 error
+	}
+	invocations      map[string][][]interface{}
+	invocationsMutex sync.RWMutex
+}
+
+func (fake *FakeRoomManagerClient) CreateRoom(arg1 context.Context, arg2 livekit.NodeID, arg3 *livekit.CreateRoomRequest, arg4 ...psrpc.RequestOption) (*livekit.Room, error) {
+	fake.createRoomMutex.Lock()
+	ret, specificReturn := fake.createRoomReturnsOnCall[len(fake.createRoomArgsForCall)]
+	fake.createRoomArgsForCall = append(fake.createRoomArgsForCall, struct {
+		arg1 context.Context
+		arg2 livekit.NodeID
+		arg3 *livekit.CreateRoomRequest
+		arg4 []psrpc.RequestOption
+	}{arg1, arg2, arg3, arg4})
+	stub := fake.CreateRoomStub
+	fakeReturns := fake.createRoomReturns
+	fake.recordInvocation("CreateRoom", []interface{}{arg1, arg2, arg3, arg4})
+	fake.createRoomMutex.Unlock()
+	if stub != nil {
+		return stub(arg1, arg2, arg3, arg4...)
+	}
+	if specificReturn {
+		return ret.result1, ret.result2
+	}
+	return fakeReturns.result1, fakeReturns.result2
+}
+
+func (fake *FakeRoomManagerClient) CreateRoomCallCount() int {
+	fake.createRoomMutex.RLock()
+	defer fake.createRoomMutex.RUnlock()
+	return len(fake.createRoomArgsForCall)
+}
+
+func (fake *FakeRoomManagerClient) CreateRoomCalls(stub func(context.Context, livekit.NodeID, *livekit.CreateRoomRequest, ...psrpc.RequestOption) (*livekit.Room, error)) {
+	fake.createRoomMutex.Lock()
+	defer fake.createRoomMutex.Unlock()
+	fake.CreateRoomStub = stub
+}
+
+func (fake *FakeRoomManagerClient) CreateRoomArgsForCall(i int) (context.Context, livekit.NodeID, *livekit.CreateRoomRequest, []psrpc.RequestOption) {
+	fake.createRoomMutex.RLock()
+	defer fake.createRoomMutex.RUnlock()
+	argsForCall := fake.createRoomArgsForCall[i]
+	return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3, argsForCall.arg4
+}
+
+func (fake *FakeRoomManagerClient) CreateRoomReturns(result1 *livekit.Room, result2 error) {
+	fake.createRoomMutex.Lock()
+	defer fake.createRoomMutex.Unlock()
+	fake.CreateRoomStub = nil
+	fake.createRoomReturns = struct {
+		result1 *livekit.Room
+		result2 error
+	}{result1, result2}
+}
+
+func (fake *FakeRoomManagerClient) CreateRoomReturnsOnCall(i int, result1 *livekit.Room, result2 error) {
+	fake.createRoomMutex.Lock()
+	defer fake.createRoomMutex.Unlock()
+	fake.CreateRoomStub = nil
+	if fake.createRoomReturnsOnCall == nil {
+		fake.createRoomReturnsOnCall = make(map[int]struct {
+			result1 *livekit.Room
+			result2 error
+		})
+	}
+	fake.createRoomReturnsOnCall[i] = struct {
+		result1 *livekit.Room
+		result2 error
+	}{result1, result2}
+}
+
+func (fake *FakeRoomManagerClient) Invocations() map[string][][]interface{} {
+	fake.invocationsMutex.RLock()
+	defer fake.invocationsMutex.RUnlock()
+	fake.createRoomMutex.RLock()
+	defer fake.createRoomMutex.RUnlock()
+	copiedInvocations := map[string][][]interface{}{}
+	for key, value := range fake.invocations {
+		copiedInvocations[key] = value
+	}
+	return copiedInvocations
+}
+
+func (fake *FakeRoomManagerClient) recordInvocation(key string, args []interface{}) {
+	fake.invocationsMutex.Lock()
+	defer fake.invocationsMutex.Unlock()
+	if fake.invocations == nil {
+		fake.invocations = map[string][][]interface{}{}
+	}
+	if fake.invocations[key] == nil {
+		fake.invocations[key] = [][]interface{}{}
+	}
+	fake.invocations[key] = append(fake.invocations[key], args)
+}
+
+var _ routing.RoomManagerClient = new(FakeRoomManagerClient)

pkg/routing/routingfakes/fake_router.go

@@ -22,6 +22,20 @@ type FakeRouter struct {
 	clearRoomStateReturnsOnCall map[int]struct {
 		result1 error
 	}
+	CreateRoomStub        func(context.Context, *livekit.CreateRoomRequest) (*livekit.Room, error)
+	createRoomMutex       sync.RWMutex
+	createRoomArgsForCall []struct {
+		arg1 context.Context
+		arg2 *livekit.CreateRoomRequest
+	}
+	createRoomReturns struct {
+		result1 *livekit.Room
+		result2 error
+	}
+	createRoomReturnsOnCall map[int]struct {
+		result1 *livekit.Room
+		result2 error
+	}
 	DrainStub        func()
 	drainMutex       sync.RWMutex
 	drainArgsForCall []struct {
@@ -200,6 +214,71 @@ func (fake *FakeRouter) ClearRoomStateReturnsOnCall(i int, result1 error) {
 	}{result1}
 }
 
+func (fake *FakeRouter) CreateRoom(arg1 context.Context, arg2 *livekit.CreateRoomRequest) (*livekit.Room, error) {
+	fake.createRoomMutex.Lock()
+	ret, specificReturn := fake.createRoomReturnsOnCall[len(fake.createRoomArgsForCall)]
+	fake.createRoomArgsForCall = append(fake.createRoomArgsForCall, struct {
+		arg1 context.Context
+		arg2 *livekit.CreateRoomRequest
+	}{arg1, arg2})
+	stub := fake.CreateRoomStub
+	fakeReturns := fake.createRoomReturns
+	fake.recordInvocation("CreateRoom", []interface{}{arg1, arg2})
+	fake.createRoomMutex.Unlock()
+	if stub != nil {
+		return stub(arg1, arg2)
+	}
+	if specificReturn {
+		return ret.result1, ret.result2
+	}
+	return fakeReturns.result1, fakeReturns.result2
+}
+
+func (fake *FakeRouter) CreateRoomCallCount() int {
+	fake.createRoomMutex.RLock()
+	defer fake.createRoomMutex.RUnlock()
+	return len(fake.createRoomArgsForCall)
+}
+
+func (fake *FakeRouter) CreateRoomCalls(stub func(context.Context, *livekit.CreateRoomRequest) (*livekit.Room, error)) {
+	fake.createRoomMutex.Lock()
+	defer fake.createRoomMutex.Unlock()
+	fake.CreateRoomStub = stub
+}
+
+func (fake *FakeRouter) CreateRoomArgsForCall(i int) (context.Context, *livekit.CreateRoomRequest) {
+	fake.createRoomMutex.RLock()
+	defer fake.createRoomMutex.RUnlock()
+	argsForCall := fake.createRoomArgsForCall[i]
+	return argsForCall.arg1, argsForCall.arg2
+}
+
+func (fake *FakeRouter) CreateRoomReturns(result1 *livekit.Room, result2 error) {
+	fake.createRoomMutex.Lock()
+	defer fake.createRoomMutex.Unlock()
+	fake.CreateRoomStub = nil
+	fake.createRoomReturns = struct {
+		result1 *livekit.Room
+		result2 error
+	}{result1, result2}
+}
+
+func (fake *FakeRouter) CreateRoomReturnsOnCall(i int, result1 *livekit.Room, result2 error) {
+	fake.createRoomMutex.Lock()
+	defer fake.createRoomMutex.Unlock()
+	fake.CreateRoomStub = nil
+	if fake.createRoomReturnsOnCall == nil {
+		fake.createRoomReturnsOnCall = make(map[int]struct {
+			result1 *livekit.Room
+			result2 error
+		})
+	}
+	fake.createRoomReturnsOnCall[i] = struct {
+		result1 *livekit.Room
+		result2 error
+	}{result1, result2}
+}
+
 func (fake *FakeRouter) Drain() {
 	fake.drainMutex.Lock()
 	fake.drainArgsForCall = append(fake.drainArgsForCall, struct {
@@ -768,6 +847,8 @@ func (fake *FakeRouter) Invocations() map[string][][]interface{} {
 	defer fake.invocationsMutex.RUnlock()
 	fake.clearRoomStateMutex.RLock()
 	defer fake.clearRoomStateMutex.RUnlock()
+	fake.createRoomMutex.RLock()
+	defer fake.createRoomMutex.RUnlock()
 	fake.drainMutex.RLock()
 	defer fake.drainMutex.RUnlock()
 	fake.getNodeForRoomMutex.RLock()

pkg/service/interfaces.go

@@ -73,7 +73,9 @@ type IngressStore interface {
 
 //counterfeiter:generate . RoomAllocator
 type RoomAllocator interface {
-	CreateRoom(ctx context.Context, req *livekit.CreateRoomRequest) (*livekit.Room, bool, error)
+	CreateRoomEnabled() bool
+	SelectRoomNode(ctx context.Context, roomName livekit.RoomName, nodeID livekit.NodeID) error
+	CreateRoom(ctx context.Context, req *livekit.CreateRoomRequest) (*livekit.Room, *livekit.RoomInternal, bool, error)
 	ValidateCreateRoom(ctx context.Context, roomName livekit.RoomName) error
 }
 

pkg/service/roomallocator.go

@@ -53,12 +53,16 @@ func NewRoomAllocator(conf *config.Config, router routing.Router, rs ObjectStore
 	}, nil
 }
 
+func (r *StandardRoomAllocator) CreateRoomEnabled() bool {
+	return r.config.Room.CreateRoomEnabled
+}
+
 // CreateRoom creates a new room from a request and allocates it to a node to handle
 // it'll also monitor its state, and cleans it up when appropriate
-func (r *StandardRoomAllocator) CreateRoom(ctx context.Context, req *livekit.CreateRoomRequest) (*livekit.Room, bool, error) {
+func (r *StandardRoomAllocator) CreateRoom(ctx context.Context, req *livekit.CreateRoomRequest) (*livekit.Room, *livekit.RoomInternal, bool, error) {
 	token, err := r.roomStore.LockRoom(ctx, livekit.RoomName(req.Name), 5*time.Second)
 	if err != nil {
-		return nil, false, err
+		return nil, nil, false, err
 	}
 	defer func() {
 		_ = r.roomStore.UnlockRoom(ctx, livekit.RoomName(req.Name), token)
@@ -78,12 +82,12 @@ func (r *StandardRoomAllocator) CreateRoom(ctx context.Context, req *livekit.Cre
 		internal = &livekit.RoomInternal{}
 		applyDefaultRoomConfig(rm, internal, &r.config.Room)
 	} else if err != nil {
-		return nil, false, err
+		return nil, nil, false, err
 	}
 
 	req, err = r.applyNamedRoomConfiguration(req)
 	if err != nil {
-		return nil, false, err
+		return nil, nil, false, err
 	}
 
 	if req.EmptyTimeout > 0 {
@@ -121,48 +125,51 @@ func (r *StandardRoomAllocator) CreateRoom(ctx context.Context, req *livekit.Cre
 	}
 
 	if err = r.roomStore.StoreRoom(ctx, rm, internal); err != nil {
-		return nil, false, err
+		return nil, nil, false, err
 	}
 
+	return rm, internal, created, nil
+}
+
+func (r *StandardRoomAllocator) SelectRoomNode(ctx context.Context, roomName livekit.RoomName, nodeID livekit.NodeID) error {
 	// check if room already assigned
-	existing, err := r.router.GetNodeForRoom(ctx, livekit.RoomName(rm.Name))
+	existing, err := r.router.GetNodeForRoom(ctx, roomName)
 	if !errors.Is(err, routing.ErrNotFound) && err != nil {
-		return nil, false, err
+		return err
 	}
 
 	// if already assigned and still available, keep it on that node
 	if err == nil && selector.IsAvailable(existing) {
 		// if node hosting the room is full, deny entry
 		if selector.LimitsReached(r.config.Limit, existing.Stats) {
-			return nil, false, routing.ErrNodeLimitReached
+			return routing.ErrNodeLimitReached
 		}
 
-		return rm, created, nil
+		return nil
 	}
 
 	// select a new node
-	nodeID := livekit.NodeID(req.NodeId)
 	if nodeID == "" {
 		nodes, err := r.router.ListNodes()
 		if err != nil {
-			return nil, false, err
+			return err
 		}
 
 		node, err := r.selector.SelectNode(nodes)
 		if err != nil {
-			return nil, false, err
+			return err
 		}
 
 		nodeID = livekit.NodeID(node.Id)
 	}
 
-	logger.Infow("selected node for room", "room", rm.Name, "roomID", rm.Sid, "selectedNodeID", nodeID)
-	err = r.router.SetNodeForRoom(ctx, livekit.RoomName(rm.Name), nodeID)
+	logger.Infow("selected node for room", "room", roomName, "selectedNodeID", nodeID)
+	err = r.router.SetNodeForRoom(ctx, roomName, nodeID)
 	if err != nil {
-		return nil, false, err
+		return err
 	}
 
-	return rm, true, nil
+	return nil
 }
 
 func (r *StandardRoomAllocator) ValidateCreateRoom(ctx context.Context, roomName livekit.RoomName) error {

pkg/service/roomallocator_test.go

@@ -39,13 +39,15 @@ func TestCreateRoom(t *testing.T) {
 
 		ra, conf := newTestRoomAllocator(t, conf, node)
 
-		room, _, err := ra.CreateRoom(context.Background(), &livekit.CreateRoomRequest{Name: "myroom"})
+		room, _, _, err := ra.CreateRoom(context.Background(), &livekit.CreateRoomRequest{Name: "myroom"})
 		require.NoError(t, err)
 		require.Equal(t, conf.Room.EmptyTimeout, room.EmptyTimeout)
 		require.Equal(t, conf.Room.DepartureTimeout, room.DepartureTimeout)
 		require.NotEmpty(t, room.EnabledCodecs)
 	})
+}
 
+func SelectRoomNode(t *testing.T) {
 	t.Run("reject new participants when track limit has been reached", func(t *testing.T) {
 		conf, err := config.NewConfig("", true, nil, nil)
 		require.NoError(t, err)
@@ -58,7 +60,7 @@ func TestCreateRoom(t *testing.T) {
 
 		ra, _ := newTestRoomAllocator(t, conf, node)
 
-		_, _, err = ra.CreateRoom(context.Background(), &livekit.CreateRoomRequest{Name: "low-limit-room"})
+		err = ra.SelectRoomNode(context.Background(), "low-limit-room", "")
 		require.ErrorIs(t, err, routing.ErrNodeLimitReached)
 	})
 
@@ -74,7 +76,7 @@ func TestCreateRoom(t *testing.T) {
 
 		ra, _ := newTestRoomAllocator(t, conf, node)
 
-		_, _, err = ra.CreateRoom(context.Background(), &livekit.CreateRoomRequest{Name: "low-limit-room"})
+		err = ra.SelectRoomNode(context.Background(), "low-limit-room", "")
 		require.ErrorIs(t, err, routing.ErrNodeLimitReached)
 	})
 }

pkg/service/roommanager.go

@@ -36,6 +36,7 @@ import (
 	"github.com/livekit/protocol/utils/guid"
 	"github.com/livekit/protocol/utils/must"
 	"github.com/livekit/psrpc"
+	"github.com/livekit/psrpc/pkg/middleware"
 
 	"github.com/livekit/livekit-server/pkg/clientconfiguration"
 	"github.com/livekit/livekit-server/pkg/config"
@@ -70,6 +71,8 @@ type RoomManager struct {
 	serverInfo        *livekit.ServerInfo
 	currentNode       routing.LocalNode
 	router            routing.Router
+	roomAllocator     RoomAllocator
+	roomManagerServer rpc.TypedRoomManagerServer
 	roomStore         ObjectStore
 	telemetry         telemetry.TelemetryService
 	clientConfManager clientconfiguration.ClientConfigurationManager
@@ -96,6 +99,7 @@ func NewLocalRoomManager(
 	roomStore ObjectStore,
 	currentNode routing.LocalNode,
 	router routing.Router,
+	roomAllocator RoomAllocator,
 	telemetry telemetry.TelemetryService,
 	clientConfManager clientconfiguration.ClientConfigurationManager,
 	agentClient agent.Client,
@@ -111,11 +115,12 @@ func NewLocalRoomManager(
 		return nil, err
 	}
 
-	return &RoomManager{
+	r := &RoomManager{
 		config:            conf,
 		rtcConfig:         rtcConf,
 		currentNode:       currentNode,
 		router:            router,
+		roomAllocator:     roomAllocator,
 		roomStore:         roomStore,
 		telemetry:         telemetry,
 		clientConfManager: clientConfManager,
@@ -139,7 +144,17 @@ func NewLocalRoomManager(
 			Region:        conf.Region,
 			NodeId:        currentNode.Id,
 		},
-	}, nil
+	}
+
+	r.roomManagerServer, err = rpc.NewTypedRoomManagerServer(r, bus, rpc.WithServerLogger(logger.GetLogger()), middleware.WithServerMetrics(rpc.PSRPCMetricsObserver{}), psrpc.WithServerChannelSize(conf.PSRPC.BufferSize))
+	if err != nil {
+		return nil, err
+	}
+	if err := r.roomManagerServer.RegisterAllNodeTopics(livekit.NodeID(currentNode.Id)); err != nil {
+		return nil, err
+	}
+
+	return r, nil
 }
 
 func (r *RoomManager) GetRoom(_ context.Context, roomName livekit.RoomName) *rtc.Room {
@@ -229,6 +244,7 @@ func (r *RoomManager) Stop() {
 		room.Close(types.ParticipantCloseReasonRoomManagerStop)
 	}
 
+	r.roomManagerServer.Kill()
 	r.roomServers.Kill()
 	r.agentDispatchServers.Kill()
 	r.participantServers.Kill()
@@ -249,17 +265,27 @@ func (r *RoomManager) Stop() {
 	}
 }
 
+func (r *RoomManager) CreateRoom(ctx context.Context, req *livekit.CreateRoomRequest) (*livekit.Room, error) {
+	room, err := r.getOrCreateRoom(ctx, req)
+	if err != nil {
+		return nil, err
+	}
+	defer room.Release()
+
+	return room.ToProto(), nil
+}
+
 // StartSession starts WebRTC session when a new participant is connected, takes place on RTC node
 func (r *RoomManager) StartSession(
 	ctx context.Context,
-	roomName livekit.RoomName,
+	createRoom *livekit.CreateRoomRequest,
 	pi routing.ParticipantInit,
 	requestSource routing.MessageSource,
 	responseSink routing.MessageSink,
 ) error {
 	sessionStartTime := time.Now()
 
-	room, err := r.getOrCreateRoom(ctx, roomName)
+	room, err := r.getOrCreateRoom(ctx, createRoom)
 	if err != nil {
 		return err
 	}
@@ -288,7 +314,7 @@ func (r *RoomManager) StartSession(
 				//
 				// It is possible that the client did not get that send request. So, send it again.
 				logger.Infow("cannot restart a closed participant",
-					"room", roomName,
+					"room", room.Name(),
 					"nodeID", r.currentNode.Id,
 					"participant", pi.Identity,
 					"reason", pi.ReconnectReason,
@@ -320,7 +346,7 @@ func (r *RoomManager) StartSession(
 				"reason", pi.ReconnectReason,
 				"numParticipants", room.GetParticipantCount(),
 			)
-			iceConfig := r.getIceConfig(roomName, participant)
+			iceConfig := r.getIceConfig(room.Name(), participant)
 			if err = room.ResumeParticipant(
 				participant,
 				requestSource,
@@ -369,7 +395,7 @@ func (r *RoomManager) StartSession(
 	}
 
 	logger.Debugw("starting RTC session",
-		"room", roomName,
+		"room", room.Name(),
 		"nodeID", r.currentNode.Id,
 		"participant", pi.Identity,
 		"clientInfo", logger.Proto(pi.Client),
@@ -464,7 +490,7 @@ func (r *RoomManager) StartSession(
 	if err != nil {
 		return err
 	}
-	iceConfig := r.setIceConfig(roomName, participant)
+	iceConfig := r.setIceConfig(room.Name(), participant)
 
 	// join room
 	opts := rtc.ParticipantOptions{
@@ -477,7 +503,7 @@ func (r *RoomManager) StartSession(
 		return err
 	}
 
-	participantTopic := rpc.FormatParticipantTopic(roomName, participant.Identity())
+	participantTopic := rpc.FormatParticipantTopic(room.Name(), participant.Identity())
 	participantServer := must.Get(rpc.NewTypedParticipantServer(r, r.bus))
 	killParticipantServer := r.participantServers.Replace(participantTopic, participantServer)
 	if err := participantServer.RegisterAllParticipantTopics(participantTopic); err != nil {
@@ -487,7 +513,7 @@ func (r *RoomManager) StartSession(
 		return err
 	}
 
-	if err = r.roomStore.StoreParticipant(ctx, roomName, participant.ToProto()); err != nil {
+	if err = r.roomStore.StoreParticipant(ctx, room.Name(), participant.ToProto()); err != nil {
 		pLogger.Errorw("could not store participant", err)
 	}
 
@@ -508,7 +534,7 @@ func (r *RoomManager) StartSession(
 	participant.OnClose(func(p types.LocalParticipant) {
 		killParticipantServer()
 
-		if err := r.roomStore.DeleteParticipant(ctx, roomName, p.Identity()); err != nil {
+		if err := r.roomStore.DeleteParticipant(ctx, room.Name(), p.Identity()); err != nil {
 			pLogger.Errorw("could not delete participant", err)
 		}
 
@@ -524,7 +550,7 @@ func (r *RoomManager) StartSession(
 		}
 	})
 	participant.OnICEConfigChanged(func(participant types.LocalParticipant, iceConfig *livekit.ICEConfig) {
-		r.iceConfigCache.Put(iceConfigCacheKey{roomName, participant.Identity()}, iceConfig)
+		r.iceConfigCache.Put(iceConfigCacheKey{room.Name(), participant.Identity()}, iceConfig)
 	})
 
 	go r.rtcSessionWorker(room, participant, requestSource)
@@ -532,7 +558,9 @@ func (r *RoomManager) StartSession(
 }
 
 // create the actual room object, to be used on RTC node
-func (r *RoomManager) getOrCreateRoom(ctx context.Context, roomName livekit.RoomName) (*rtc.Room, error) {
+func (r *RoomManager) getOrCreateRoom(ctx context.Context, createRoom *livekit.CreateRoomRequest) (*rtc.Room, error) {
+	roomName := livekit.RoomName(createRoom.Name)
+
 	r.lock.RLock()
 	lastSeenRoom := r.rooms[roomName]
 	r.lock.RUnlock()
@@ -542,7 +570,7 @@ func (r *RoomManager) getOrCreateRoom(ctx context.Context, roomName livekit.Room
 	}
 
 	// create new room, get details first
-	ri, internal, err := r.roomStore.LoadRoom(ctx, roomName, true)
+	ri, internal, created, err := r.roomAllocator.CreateRoom(ctx, createRoom)
 	if err != nil {
 		return nil, err
 	}
@@ -617,6 +645,21 @@ func (r *RoomManager) getOrCreateRoom(ctx context.Context, roomName livekit.Room
 	r.telemetry.RoomStarted(ctx, newRoom.ToProto())
 	prometheus.RoomStarted()
 
+	if created && createRoom.GetEgress().GetRoom() != nil {
+		// ensure room name matches
+		createRoom.Egress.Room.RoomName = createRoom.Name
+		_, err = r.egressLauncher.StartEgress(ctx, &rpc.StartEgressRequest{
+			Request: &rpc.StartEgressRequest_RoomComposite{
+				RoomComposite: createRoom.Egress.Room,
+			},
+			RoomId: ri.Sid,
+		})
+		if err != nil {
+			newRoom.Release()
+			return nil, err
+		}
+	}
+
 	return newRoom, nil
 }
 

pkg/service/roomservice.go

@@ -36,7 +36,6 @@ import (
 type RoomService struct {
 	limitConf         config.LimitConfig
 	apiConf           config.APIConfig
-	psrpcConf         rpc.PSRPCConfig
 	router            routing.MessageRouter
 	roomAllocator     RoomAllocator
 	roomStore         ServiceStore
@@ -50,7 +49,6 @@ type RoomService struct {
 func NewRoomService(
 	limitConf config.LimitConfig,
 	apiConf config.APIConfig,
-	psrpcConf rpc.PSRPCConfig,
 	router routing.MessageRouter,
 	roomAllocator RoomAllocator,
 	serviceStore ServiceStore,
@@ -63,7 +61,6 @@ func NewRoomService(
 	svc = &RoomService{
 		limitConf:         limitConf,
 		apiConf:           apiConf,
-		psrpcConf:         psrpcConf,
 		router:            router,
 		roomAllocator:     roomAllocator,
 		roomStore:         serviceStore,
@@ -88,11 +85,24 @@ func (s *RoomService) CreateRoom(ctx context.Context, req *livekit.CreateRoomReq
 		return nil, fmt.Errorf("%w: max length %d", ErrRoomNameExceedsLimits, s.limitConf.MaxRoomNameLength)
 	}
 
-	rm, created, err := s.roomAllocator.CreateRoom(ctx, req)
+	if s.roomAllocator.CreateRoomEnabled() {
+		err := s.roomAllocator.SelectRoomNode(ctx, livekit.RoomName(req.Name), livekit.NodeID(req.NodeId))
+		if err != nil {
+			return nil, err
+		}
+
+		return s.router.CreateRoom(ctx, req)
+	}
+
+	rm, _, created, err := s.roomAllocator.CreateRoom(ctx, req)
 	if err != nil {
 		err = errors.Wrap(err, "could not create room")
 		return nil, err
 	}
+	err = s.roomAllocator.SelectRoomNode(ctx, livekit.RoomName(req.Name), livekit.NodeID(req.NodeId))
+	if err != nil {
+		return nil, err
+	}
 
 	done, err := s.startRoom(ctx, livekit.RoomName(req.Name))
 	if err != nil {
@@ -153,11 +163,18 @@ func (s *RoomService) DeleteRoom(ctx context.Context, req *livekit.DeleteRoomReq
 		return nil, err
 	}
 
-	done, err := s.startRoom(ctx, livekit.RoomName(req.Room))
-	if err != nil {
-		return nil, err
+	if s.roomAllocator.CreateRoomEnabled() {
+		_, err := s.router.CreateRoom(ctx, &livekit.CreateRoomRequest{Name: req.Room})
+		if err != nil {
+			return nil, err
+		}
+	} else {
+		done, err := s.startRoom(ctx, livekit.RoomName(req.Room))
+		if err != nil {
+			return nil, err
+		}
+		defer done()
 	}
-	defer done()
 
 	_, err = s.roomClient.DeleteRoom(ctx, s.topicFormatter.RoomTopic(ctx, livekit.RoomName(req.Room)), req)
 	if err != nil {

pkg/service/roomservice_test.go

@@ -111,7 +111,6 @@ func newTestRoomService(limitConf config.LimitConfig) *TestRoomService {
 	svc, err := service.NewRoomService(
 		limitConf,
 		config.APIConfig{ExecutionTimeout: 2},
-		rpc.PSRPCConfig{},
 		router,
 		allocator,
 		store,

pkg/service/rtcservice.go

@@ -179,6 +179,10 @@ func (s *RTCService) validate(r *http.Request) (livekit.RoomName, routing.Partic
 		Client:          s.ParseClientInfo(r),
 		Grants:          claims,
 		Region:          region,
+		CreateRoom: &livekit.CreateRoomRequest{
+			Name:       string(roomName),
+			ConfigName: GetRoomConfiguration(r.Context()),
+		},
 	}
 	if pi.Reconnect {
 		pi.ID = livekit.ParticipantID(participantID)
@@ -214,14 +218,13 @@ func (s *RTCService) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	// for logger
-	loggerFields := []interface{}{
+	loggerFields := []any{
 		"participant", pi.Identity,
+		"pID", pi.ID,
 		"room", roomName,
 		"remote", false,
 	}
-
-	l := utils.GetLogger(r.Context())
+	pLogger := utils.GetLogger(r.Context()).WithValues(loggerFields...)
 
 	// give it a few attempts to start session
 	var cr connectionResult
@@ -234,8 +237,7 @@ func (s *RTCService) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 			break
 		}
 		if i < 2 {
-			fieldsWithAttempt := append(loggerFields, "attempt", i)
-			l.Warnw("failed to start connection, retrying", err, fieldsWithAttempt...)
+			pLogger.Warnw("failed to start connection, retrying", err, "attempt", i)
 		}
 	}
 
@@ -257,13 +259,6 @@ func (s *RTCService) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		signalStats.ResolveParticipant(join.GetParticipant())
 	}
 
-	pLogger := rtc.LoggerWithParticipant(
-		rtc.LoggerWithRoom(l, roomName, livekit.RoomID(cr.Room.Sid)),
-		pi.Identity,
-		pi.ID,
-		false,
-	)
-
 	closedByClient := atomic.NewBool(false)
 	done := make(chan struct{})
 	// function exits when websocket terminates, it'll close the event reading off of request sink and response source as well
@@ -519,8 +514,14 @@ func (s *RTCService) startConnection(
 	var cr connectionResult
 	var err error
 
-	cr.Room, _, err = s.roomAllocator.CreateRoom(ctx, &livekit.CreateRoomRequest{Name: string(roomName), ConfigName: GetRoomConfiguration(ctx)})
-	if err != nil {
+	if !s.roomAllocator.CreateRoomEnabled() {
+		cr.Room, _, _, err = s.roomAllocator.CreateRoom(ctx, pi.CreateRoom)
+		if err != nil {
+			return cr, nil, err
+		}
+	}
+
+	if err := s.roomAllocator.SelectRoomNode(ctx, roomName, ""); err != nil {
 		return cr, nil, err
 	}
 

pkg/service/servicefakes/fake_room_allocator.go

@@ -10,7 +10,7 @@ import (
 )
 
 type FakeRoomAllocator struct {
-	CreateRoomStub        func(context.Context, *livekit.CreateRoomRequest) (*livekit.Room, bool, error)
+	CreateRoomStub        func(context.Context, *livekit.CreateRoomRequest) (*livekit.Room, *livekit.RoomInternal, bool, error)
 	createRoomMutex       sync.RWMutex
 	createRoomArgsForCall []struct {
 		arg1 context.Context
@@ -18,13 +18,38 @@ type FakeRoomAllocator struct {
 	}
 	createRoomReturns struct {
 		result1 *livekit.Room
-		result2 bool
-		result3 error
+		result2 *livekit.RoomInternal
+		result3 bool
+		result4 error
 	}
 	createRoomReturnsOnCall map[int]struct {
 		result1 *livekit.Room
-		result2 bool
-		result3 error
+		result2 *livekit.RoomInternal
+		result3 bool
+		result4 error
+	}
+	SelectRoomNodeStub        func(context.Context, livekit.RoomName, livekit.NodeID) error
+	selectRoomNodeMutex       sync.RWMutex
+	selectRoomNodeArgsForCall []struct {
+		arg1 context.Context
+		arg2 livekit.RoomName
+		arg3 livekit.NodeID
+	}
+	selectRoomNodeReturns struct {
+		result1 error
+	}
+	selectRoomNodeReturnsOnCall map[int]struct {
+		result1 error
+	}
+	StartRoomEnabledStub        func() bool
+	startRoomEnabledMutex       sync.RWMutex
+	startRoomEnabledArgsForCall []struct {
+	}
+	startRoomEnabledReturns struct {
+		result1 bool
+	}
+	startRoomEnabledReturnsOnCall map[int]struct {
+		result1 bool
 	}
 	ValidateCreateRoomStub        func(context.Context, livekit.RoomName) error
 	validateCreateRoomMutex       sync.RWMutex
@@ -42,7 +67,7 @@ type FakeRoomAllocator struct {
 	invocationsMutex sync.RWMutex
 }
 
-func (fake *FakeRoomAllocator) CreateRoom(arg1 context.Context, arg2 *livekit.CreateRoomRequest) (*livekit.Room, bool, error) {
+func (fake *FakeRoomAllocator) CreateRoom(arg1 context.Context, arg2 *livekit.CreateRoomRequest) (*livekit.Room, *livekit.RoomInternal, bool, error) {
 	fake.createRoomMutex.Lock()
 	ret, specificReturn := fake.createRoomReturnsOnCall[len(fake.createRoomArgsForCall)]
 	fake.createRoomArgsForCall = append(fake.createRoomArgsForCall, struct {
@@ -57,9 +82,9 @@ func (fake *FakeRoomAllocator) CreateRoom(arg1 context.Context, arg2 *livekit.Cr
 		return stub(arg1, arg2)
 	}
 	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 *FakeRoomAllocator) CreateRoomCallCount() int {
@@ -68,7 +93,7 @@ func (fake *FakeRoomAllocator) CreateRoomCallCount() int {
 	return len(fake.createRoomArgsForCall)
 }
 
-func (fake *FakeRoomAllocator) CreateRoomCalls(stub func(context.Context, *livekit.CreateRoomRequest) (*livekit.Room, bool, error)) {
+func (fake *FakeRoomAllocator) CreateRoomCalls(stub func(context.Context, *livekit.CreateRoomRequest) (*livekit.Room, *livekit.RoomInternal, bool, error)) {
 	fake.createRoomMutex.Lock()
 	defer fake.createRoomMutex.Unlock()
 	fake.CreateRoomStub = stub
@@ -81,33 +106,152 @@ func (fake *FakeRoomAllocator) CreateRoomArgsForCall(i int) (context.Context, *l
 	return argsForCall.arg1, argsForCall.arg2
 }
 
-func (fake *FakeRoomAllocator) CreateRoomReturns(result1 *livekit.Room, result2 bool, result3 error) {
+func (fake *FakeRoomAllocator) CreateRoomReturns(result1 *livekit.Room, result2 *livekit.RoomInternal, result3 bool, result4 error) {
 	fake.createRoomMutex.Lock()
 	defer fake.createRoomMutex.Unlock()
 	fake.CreateRoomStub = nil
 	fake.createRoomReturns = struct {
 		result1 *livekit.Room
-		result2 bool
-		result3 error
-	}{result1, result2, result3}
+		result2 *livekit.RoomInternal
+		result3 bool
+		result4 error
+	}{result1, result2, result3, result4}
 }
 
-func (fake *FakeRoomAllocator) CreateRoomReturnsOnCall(i int, result1 *livekit.Room, result2 bool, result3 error) {
+func (fake *FakeRoomAllocator) CreateRoomReturnsOnCall(i int, result1 *livekit.Room, result2 *livekit.RoomInternal, result3 bool, result4 error) {
 	fake.createRoomMutex.Lock()
 	defer fake.createRoomMutex.Unlock()
 	fake.CreateRoomStub = nil
 	if fake.createRoomReturnsOnCall == nil {
 		fake.createRoomReturnsOnCall = make(map[int]struct {
 			result1 *livekit.Room
-			result2 bool
-			result3 error
+			result2 *livekit.RoomInternal
+			result3 bool
+			result4 error
 		})
 	}
 	fake.createRoomReturnsOnCall[i] = struct {
 		result1 *livekit.Room
-		result2 bool
-		result3 error
-	}{result1, result2, result3}
+		result2 *livekit.RoomInternal
+		result3 bool
+		result4 error
+	}{result1, result2, result3, result4}
+}
+
+func (fake *FakeRoomAllocator) SelectRoomNode(arg1 context.Context, arg2 livekit.RoomName, arg3 livekit.NodeID) error {
+	fake.selectRoomNodeMutex.Lock()
+	ret, specificReturn := fake.selectRoomNodeReturnsOnCall[len(fake.selectRoomNodeArgsForCall)]
+	fake.selectRoomNodeArgsForCall = append(fake.selectRoomNodeArgsForCall, struct {
+		arg1 context.Context
+		arg2 livekit.RoomName
+		arg3 livekit.NodeID
+	}{arg1, arg2, arg3})
+	stub := fake.SelectRoomNodeStub
+	fakeReturns := fake.selectRoomNodeReturns
+	fake.recordInvocation("SelectRoomNode", []interface{}{arg1, arg2, arg3})
+	fake.selectRoomNodeMutex.Unlock()
+	if stub != nil {
+		return stub(arg1, arg2, arg3)
+	}
+	if specificReturn {
+		return ret.result1
+	}
+	return fakeReturns.result1
+}
+
+func (fake *FakeRoomAllocator) SelectRoomNodeCallCount() int {
+	fake.selectRoomNodeMutex.RLock()
+	defer fake.selectRoomNodeMutex.RUnlock()
+	return len(fake.selectRoomNodeArgsForCall)
+}
+
+func (fake *FakeRoomAllocator) SelectRoomNodeCalls(stub func(context.Context, livekit.RoomName, livekit.NodeID) error) {
+	fake.selectRoomNodeMutex.Lock()
+	defer fake.selectRoomNodeMutex.Unlock()
+	fake.SelectRoomNodeStub = stub
+}
+
+func (fake *FakeRoomAllocator) SelectRoomNodeArgsForCall(i int) (context.Context, livekit.RoomName, livekit.NodeID) {
+	fake.selectRoomNodeMutex.RLock()
+	defer fake.selectRoomNodeMutex.RUnlock()
+	argsForCall := fake.selectRoomNodeArgsForCall[i]
+	return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3
+}
+
+func (fake *FakeRoomAllocator) SelectRoomNodeReturns(result1 error) {
+	fake.selectRoomNodeMutex.Lock()
+	defer fake.selectRoomNodeMutex.Unlock()
+	fake.SelectRoomNodeStub = nil
+	fake.selectRoomNodeReturns = struct {
+		result1 error
+	}{result1}
+}
+
+func (fake *FakeRoomAllocator) SelectRoomNodeReturnsOnCall(i int, result1 error) {
+	fake.selectRoomNodeMutex.Lock()
+	defer fake.selectRoomNodeMutex.Unlock()
+	fake.SelectRoomNodeStub = nil
+	if fake.selectRoomNodeReturnsOnCall == nil {
+		fake.selectRoomNodeReturnsOnCall = make(map[int]struct {
+			result1 error
+		})
+	}
+	fake.selectRoomNodeReturnsOnCall[i] = struct {
+		result1 error
+	}{result1}
+}
+
+func (fake *FakeRoomAllocator) CreateRoomEnabled() bool {
+	fake.startRoomEnabledMutex.Lock()
+	ret, specificReturn := fake.startRoomEnabledReturnsOnCall[len(fake.startRoomEnabledArgsForCall)]
+	fake.startRoomEnabledArgsForCall = append(fake.startRoomEnabledArgsForCall, struct {
+	}{})
+	stub := fake.StartRoomEnabledStub
+	fakeReturns := fake.startRoomEnabledReturns
+	fake.recordInvocation("StartRoomEnabled", []interface{}{})
+	fake.startRoomEnabledMutex.Unlock()
+	if stub != nil {
+		return stub()
+	}
+	if specificReturn {
+		return ret.result1
+	}
+	return fakeReturns.result1
+}
+
+func (fake *FakeRoomAllocator) StartRoomEnabledCallCount() int {
+	fake.startRoomEnabledMutex.RLock()
+	defer fake.startRoomEnabledMutex.RUnlock()
+	return len(fake.startRoomEnabledArgsForCall)
+}
+
+func (fake *FakeRoomAllocator) StartRoomEnabledCalls(stub func() bool) {
+	fake.startRoomEnabledMutex.Lock()
+	defer fake.startRoomEnabledMutex.Unlock()
+	fake.StartRoomEnabledStub = stub
+}
+
+func (fake *FakeRoomAllocator) StartRoomEnabledReturns(result1 bool) {
+	fake.startRoomEnabledMutex.Lock()
+	defer fake.startRoomEnabledMutex.Unlock()
+	fake.StartRoomEnabledStub = nil
+	fake.startRoomEnabledReturns = struct {
+		result1 bool
+	}{result1}
+}
+
+func (fake *FakeRoomAllocator) StartRoomEnabledReturnsOnCall(i int, result1 bool) {
+	fake.startRoomEnabledMutex.Lock()
+	defer fake.startRoomEnabledMutex.Unlock()
+	fake.StartRoomEnabledStub = nil
+	if fake.startRoomEnabledReturnsOnCall == nil {
+		fake.startRoomEnabledReturnsOnCall = make(map[int]struct {
+			result1 bool
+		})
+	}
+	fake.startRoomEnabledReturnsOnCall[i] = struct {
+		result1 bool
+	}{result1}
 }
 
 func (fake *FakeRoomAllocator) ValidateCreateRoom(arg1 context.Context, arg2 livekit.RoomName) error {
@@ -177,6 +321,10 @@ func (fake *FakeRoomAllocator) Invocations() map[string][][]interface{} {
 	defer fake.invocationsMutex.RUnlock()
 	fake.createRoomMutex.RLock()
 	defer fake.createRoomMutex.RUnlock()
+	fake.selectRoomNodeMutex.RLock()
+	defer fake.selectRoomNodeMutex.RUnlock()
+	fake.startRoomEnabledMutex.RLock()
+	defer fake.startRoomEnabledMutex.RUnlock()
 	fake.validateCreateRoomMutex.RLock()
 	defer fake.validateCreateRoomMutex.RUnlock()
 	copiedInvocations := map[string][][]interface{}{}

pkg/service/servicefakes/fake_session_handler.go

@@ -12,11 +12,11 @@ import (
 )
 
 type FakeSessionHandler struct {
-	HandleSessionStub        func(context.Context, livekit.RoomName, routing.ParticipantInit, livekit.ConnectionID, routing.MessageSource, routing.MessageSink) error
+	HandleSessionStub        func(context.Context, *livekit.CreateRoomRequest, routing.ParticipantInit, livekit.ConnectionID, routing.MessageSource, routing.MessageSink) error
 	handleSessionMutex       sync.RWMutex
 	handleSessionArgsForCall []struct {
 		arg1 context.Context
-		arg2 livekit.RoomName
+		arg2 *livekit.CreateRoomRequest
 		arg3 routing.ParticipantInit
 		arg4 livekit.ConnectionID
 		arg5 routing.MessageSource
@@ -43,12 +43,12 @@ type FakeSessionHandler struct {
 	invocationsMutex sync.RWMutex
 }
 
-func (fake *FakeSessionHandler) HandleSession(arg1 context.Context, arg2 livekit.RoomName, arg3 routing.ParticipantInit, arg4 livekit.ConnectionID, arg5 routing.MessageSource, arg6 routing.MessageSink) error {
+func (fake *FakeSessionHandler) HandleSession(arg1 context.Context, arg2 *livekit.CreateRoomRequest, arg3 routing.ParticipantInit, arg4 livekit.ConnectionID, arg5 routing.MessageSource, arg6 routing.MessageSink) error {
 	fake.handleSessionMutex.Lock()
 	ret, specificReturn := fake.handleSessionReturnsOnCall[len(fake.handleSessionArgsForCall)]
 	fake.handleSessionArgsForCall = append(fake.handleSessionArgsForCall, struct {
 		arg1 context.Context
-		arg2 livekit.RoomName
+		arg2 *livekit.CreateRoomRequest
 		arg3 routing.ParticipantInit
 		arg4 livekit.ConnectionID
 		arg5 routing.MessageSource
@@ -73,13 +73,13 @@ func (fake *FakeSessionHandler) HandleSessionCallCount() int {
 	return len(fake.handleSessionArgsForCall)
 }
 
-func (fake *FakeSessionHandler) HandleSessionCalls(stub func(context.Context, livekit.RoomName, routing.ParticipantInit, livekit.ConnectionID, routing.MessageSource, routing.MessageSink) error) {
+func (fake *FakeSessionHandler) HandleSessionCalls(stub func(context.Context, *livekit.CreateRoomRequest, routing.ParticipantInit, livekit.ConnectionID, routing.MessageSource, routing.MessageSink) error) {
 	fake.handleSessionMutex.Lock()
 	defer fake.handleSessionMutex.Unlock()
 	fake.HandleSessionStub = stub
 }
 
-func (fake *FakeSessionHandler) HandleSessionArgsForCall(i int) (context.Context, livekit.RoomName, routing.ParticipantInit, livekit.ConnectionID, routing.MessageSource, routing.MessageSink) {
+func (fake *FakeSessionHandler) HandleSessionArgsForCall(i int) (context.Context, *livekit.CreateRoomRequest, routing.ParticipantInit, livekit.ConnectionID, routing.MessageSource, routing.MessageSink) {
 	fake.handleSessionMutex.RLock()
 	defer fake.handleSessionMutex.RUnlock()
 	argsForCall := fake.handleSessionArgsForCall[i]

pkg/service/signal.go

@@ -39,7 +39,7 @@ type SessionHandler interface {
 
 	HandleSession(
 		ctx context.Context,
-		roomName livekit.RoomName,
+		createRoom *livekit.CreateRoomRequest,
 		pi routing.ParticipantInit,
 		connectionID livekit.ConnectionID,
 		requestSource routing.MessageSource,
@@ -94,7 +94,7 @@ func (s *defaultSessionHandler) Logger(ctx context.Context) logger.Logger {
 
 func (s *defaultSessionHandler) HandleSession(
 	ctx context.Context,
-	roomName livekit.RoomName,
+	createRoom *livekit.CreateRoomRequest,
 	pi routing.ParticipantInit,
 	connectionID livekit.ConnectionID,
 	requestSource routing.MessageSource,
@@ -102,7 +102,7 @@ func (s *defaultSessionHandler) HandleSession(
 ) error {
 	prometheus.IncrementParticipantRtcInit(1)
 
-	rtcNode, err := s.router.GetNodeForRoom(ctx, roomName)
+	rtcNode, err := s.router.GetNodeForRoom(ctx, livekit.RoomName(createRoom.Name))
 	if err != nil {
 		return err
 	}
@@ -115,12 +115,12 @@ func (s *defaultSessionHandler) HandleSession(
 		return err
 	}
 
-	return s.roomManager.StartSession(ctx, roomName, pi, requestSource, responseSink)
+	return s.roomManager.StartSession(ctx, createRoom, pi, requestSource, responseSink)
 }
 
 func (s *SignalServer) Start() error {
 	logger.Debugw("starting relay signal server", "topic", s.nodeID)
-	return s.server.RegisterRelaySignalTopic(s.nodeID)
+	return s.server.RegisterAllNodeTopics(s.nodeID)
 }
 
 func (r *SignalServer) Stop() {
@@ -182,7 +182,18 @@ func (r *signalService) RelaySignal(stream psrpc.ServerStream[*rpc.RelaySignalRe
 	// copy the incoming rpc headers to avoid dropping any session vars.
 	ctx := metadata.NewContextWithIncomingHeader(context.Background(), metadata.IncomingHeader(stream.Context()))
 
-	err = r.sessionHandler.HandleSession(ctx, livekit.RoomName(ss.RoomName), *pi, livekit.ConnectionID(ss.ConnectionId), reqChan, sink)
+	createRoom := ss.CreateRoom
+	if createRoom == nil {
+		createRoom = &livekit.CreateRoomRequest{
+			Name: ss.RoomName,
+		}
+
+		if pi.Grants != nil && pi.Grants.Video != nil {
+			createRoom.ConfigName = pi.Grants.Video.RoomConfiguration
+		}
+	}
+
+	err = r.sessionHandler.HandleSession(ctx, createRoom, *pi, livekit.ConnectionID(ss.ConnectionId), reqChan, sink)
 	if err != nil {
 		sink.Close()
 		l.Errorw("could not handle new participant", err)

pkg/service/signal_test.go

@@ -67,7 +67,7 @@ func TestSignal(t *testing.T) {
 			LoggerStub: func(context.Context) logger.Logger { return logger.GetLogger() },
 			HandleSessionStub: func(
 				ctx context.Context,
-				roomName livekit.RoomName,
+				createRoom *livekit.CreateRoomRequest,
 				pi routing.ParticipantInit,
 				connectionID livekit.ConnectionID,
 				requestSource routing.MessageSource,
@@ -124,7 +124,7 @@ func TestSignal(t *testing.T) {
 			LoggerStub: func(context.Context) logger.Logger { return logger.GetLogger() },
 			HandleSessionStub: func(
 				ctx context.Context,
-				roomName livekit.RoomName,
+				createRoom *livekit.CreateRoomRequest,
 				pi routing.ParticipantInit,
 				connectionID livekit.ConnectionID,
 				requestSource routing.MessageSource,

pkg/service/wire.go

@@ -85,6 +85,8 @@ func InitializeServer(conf *config.Config, currentNode routing.LocalNode) (*Live
 		getSignalRelayConfig,
 		NewDefaultSignalServer,
 		routing.NewSignalClient,
+		getRoomConfig,
+		routing.NewRoomManagerClient,
 		rpc.NewKeepalivePubSub,
 		getPSRPCConfig,
 		getPSRPCClientParams,
@@ -111,6 +113,8 @@ func InitializeRouter(conf *config.Config, currentNode routing.LocalNode) (routi
 		getPSRPCConfig,
 		getPSRPCClientParams,
 		routing.NewSignalClient,
+		getRoomConfig,
+		routing.NewRoomManagerClient,
 		rpc.NewKeepalivePubSub,
 		routing.CreateRouter,
 	)
@@ -239,6 +243,10 @@ func getLimitConf(config *config.Config) config.LimitConfig {
 	return config.Limit
 }
 
+func getRoomConfig(config *config.Config) config.RoomConfig {
+	return config.Room
+}
+
 func getSignalRelayConfig(config *config.Config) config.SignalRelayConfig {
 	return config.SignalRelay
 }

pkg/service/wire_gen.go

@@ -38,7 +38,6 @@ import (
 func InitializeServer(conf *config.Config, currentNode routing.LocalNode) (*LivekitServer, error) {
 	limitConfig := getLimitConf(conf)
 	apiConfig := config.DefaultAPIConfig()
-	psrpcConfig := getPSRPCConfig(conf)
 	universalClient, err := createRedisClient(conf)
 	if err != nil {
 		return nil, err
@@ -50,12 +49,18 @@ func InitializeServer(conf *config.Config, currentNode routing.LocalNode) (*Live
 	if err != nil {
 		return nil, err
 	}
+	psrpcConfig := getPSRPCConfig(conf)
 	clientParams := getPSRPCClientParams(psrpcConfig, messageBus)
+	roomConfig := getRoomConfig(conf)
+	roomManagerClient, err := routing.NewRoomManagerClient(clientParams, roomConfig)
+	if err != nil {
+		return nil, err
+	}
 	keepalivePubSub, err := rpc.NewKeepalivePubSub(clientParams)
 	if err != nil {
 		return nil, err
 	}
-	router := routing.CreateRouter(universalClient, currentNode, signalClient, keepalivePubSub)
+	router := routing.CreateRouter(universalClient, currentNode, signalClient, roomManagerClient, keepalivePubSub)
 	objectStore := createStore(universalClient)
 	roomAllocator, err := NewRoomAllocator(conf, router, objectStore)
 	if err != nil {
@@ -96,7 +101,7 @@ func InitializeServer(conf *config.Config, currentNode routing.LocalNode) (*Live
 	if err != nil {
 		return nil, err
 	}
-	roomService, err := NewRoomService(limitConfig, apiConfig, psrpcConfig, router, roomAllocator, objectStore, client, rtcEgressLauncher, topicFormatter, roomClient, participantClient)
+	roomService, err := NewRoomService(limitConfig, apiConfig, router, roomAllocator, objectStore, client, rtcEgressLauncher, topicFormatter, roomClient, participantClient)
 	if err != nil {
 		return nil, err
 	}
@@ -128,7 +133,7 @@ func InitializeServer(conf *config.Config, currentNode routing.LocalNode) (*Live
 	timedVersionGenerator := utils.NewDefaultTimedVersionGenerator()
 	turnAuthHandler := NewTURNAuthHandler(keyProvider)
 	forwardStats := createForwardStats(conf)
-	roomManager, err := NewLocalRoomManager(conf, objectStore, currentNode, router, telemetryService, clientConfigurationManager, client, agentStore, rtcEgressLauncher, timedVersionGenerator, turnAuthHandler, messageBus, forwardStats)
+	roomManager, err := NewLocalRoomManager(conf, objectStore, currentNode, router, roomAllocator, telemetryService, clientConfigurationManager, client, agentStore, rtcEgressLauncher, timedVersionGenerator, turnAuthHandler, messageBus, forwardStats)
 	if err != nil {
 		return nil, err
 	}
@@ -162,11 +167,16 @@ func InitializeRouter(conf *config.Config, currentNode routing.LocalNode) (routi
 	}
 	psrpcConfig := getPSRPCConfig(conf)
 	clientParams := getPSRPCClientParams(psrpcConfig, messageBus)
+	roomConfig := getRoomConfig(conf)
+	roomManagerClient, err := routing.NewRoomManagerClient(clientParams, roomConfig)
+	if err != nil {
+		return nil, err
+	}
 	keepalivePubSub, err := rpc.NewKeepalivePubSub(clientParams)
 	if err != nil {
 		return nil, err
 	}
-	router := routing.CreateRouter(universalClient, currentNode, signalClient, keepalivePubSub)
+	router := routing.CreateRouter(universalClient, currentNode, signalClient, roomManagerClient, keepalivePubSub)
 	return router, nil
 }
 
@@ -293,6 +303,10 @@ func getLimitConf(config2 *config.Config) config.LimitConfig {
 	return config2.Limit
 }
 
+func getRoomConfig(config2 *config.Config) config.RoomConfig {
+	return config2.Room
+}
+
 func getSignalRelayConfig(config2 *config.Config) config.SignalRelayConfig {
 	return config2.SignalRelay
 }

test/agent_test.go

@@ -173,6 +173,7 @@ func TestAgentMultiNode(t *testing.T) {
 	ac1, err := newAgentClient(agentToken(), defaultServerPort)
 	require.NoError(t, err)
 	ac2, err := newAgentClient(agentToken(), defaultServerPort)
+	require.NoError(t, err)
 	defer ac1.close()
 	defer ac2.close()
 	ac1.Run(livekit.JobType_JT_ROOM, "default")