Ability to disable auto-create (#361)

* setting to disable autocreate

* improve test reliability

* update comment

* update to address feedback

config-sample.yaml

@@ -81,6 +81,8 @@ keys:
 # Each room created will inherit these settings. If rooms are created explicitly with CreateRoom, they will take
 # precedence over defaults
 # room:
+#   # allow rooms to be automatically created when participants join, defaults to true
+#   # auto_create: false
 #   # number of seconds to leave a room open when it's empty
 #   empty_timeout: 300
 #   # limit number of participants that can be in a room, 0 for no limit

magefile.go

@@ -118,7 +118,7 @@ func PublishDocker() error {
 // run unit tests, skipping integration
 func Test() error {
 	mg.Deps(generateWire, macULimit)
-	cmd := exec.Command("go", "test", "-short", "./...")
+	cmd := exec.Command("go", "test", "-short", "./...", "-count=1")
 	connectStd(cmd)
 	return cmd.Run()
 }
@@ -127,7 +127,7 @@ func Test() error {
 func TestAll() error {
 	mg.Deps(generateWire, macULimit)
 	// "-v", "-race",
-	cmd := exec.Command("go", "test", "./...", "-count=1", "-timeout=4m")
+	cmd := exec.Command("go", "test", "./...", "-count=1", "-timeout=4m", "-v")
 	connectStd(cmd)
 	return cmd.Run()
 }

pkg/config/config.go

@@ -103,6 +103,8 @@ type RedisConfig struct {
 }
 
 type RoomConfig struct {
+	// enable rooms to be automatically created
+	AutoCreate         bool        `yaml:"auto_create"`
 	EnabledCodecs      []CodecSpec `yaml:"enabled_codecs"`
 	MaxParticipants    uint32      `yaml:"max_participants"`
 	EmptyTimeout       uint32      `yaml:"empty_timeout"`
@@ -187,6 +189,7 @@ func NewConfig(confString string, c *cli.Context) (*Config, error) {
 		},
 		Redis: RedisConfig{},
 		Room: RoomConfig{
+			AutoCreate: true,
 			// by default only enable opus and VP8
 			EnabledCodecs: []CodecSpec{
 				{Mime: webrtc.MimeTypeOpus},

pkg/config/config_test.go

@@ -13,3 +13,12 @@ func TestConfig_UnmarshalKeys(t *testing.T) {
 	require.NoError(t, conf.unmarshalKeys("key1: secret1"))
 	require.Equal(t, "secret1", conf.Keys["key1"])
 }
+
+func TestConfig_DefaultsKept(t *testing.T) {
+	const content = `room:
+  empty_timeout: 10`
+	conf, err := NewConfig(content, nil)
+	require.NoError(t, err)
+	require.Equal(t, true, conf.Room.AutoCreate)
+	require.Equal(t, uint32(10), conf.Room.EmptyTimeout)
+}

pkg/service/errors.go

@@ -9,4 +9,5 @@ var (
 	ErrParticipantNotFound  = errors.New("participant does not exist")
 	ErrTrackNotFound        = errors.New("track is not found")
 	ErrWebHookMissingAPIKey = errors.New("api_key is required to use webhooks")
+	ErrOperationFailed      = errors.New("operation cannot be completed")
 )

pkg/service/roomservice.go

@@ -2,6 +2,7 @@ package service
 
 import (
 	"context"
+	"time"
 
 	"github.com/livekit/protocol/livekit"
 	"github.com/pkg/errors"
@@ -11,6 +12,11 @@ import (
 	"github.com/livekit/livekit-server/pkg/routing"
 )
 
+const (
+	executionTimeout = 2 * time.Second
+	checkInterval    = 50 * time.Millisecond
+)
+
 // A rooms service that supports a single node
 type RoomService struct {
 	router        routing.MessageRouter
@@ -75,6 +81,21 @@ func (s *RoomService) DeleteRoom(ctx context.Context, req *livekit.DeleteRoomReq
 		return nil, err
 	}
 
+	// we should not return until when the room is confirmed deleted
+	err = confirmExecution(func() error {
+		_, err := s.roomStore.LoadRoom(ctx, livekit.RoomName(req.Room))
+		if err == nil {
+			return ErrOperationFailed
+		} else if err != ErrRoomNotFound {
+			return err
+		} else {
+			return nil
+		}
+	})
+	if err != nil {
+		return nil, err
+	}
+
 	return &livekit.DeleteRoomResponse{}, nil
 }
 
@@ -252,3 +273,20 @@ func (s *RoomService) writeRoomMessage(ctx context.Context, room livekit.RoomNam
 
 	return s.router.WriteRoomRTC(ctx, room, identity, msg)
 }
+
+func confirmExecution(f func() error) error {
+	expired := time.After(executionTimeout)
+	var err error
+	for {
+		select {
+		case <-expired:
+			return err
+		default:
+			err = f()
+			if err == nil {
+				return nil
+			}
+			time.Sleep(checkInterval)
+		}
+	}
+}

pkg/service/roomservice_test.go

@@ -24,6 +24,7 @@ func TestDeleteRoom(t *testing.T) {
 			},
 		}
 		ctx := context.WithValue(context.Background(), grantsKey, grant)
+		svc.store.LoadRoomReturns(nil, service.ErrRoomNotFound)
 		_, err := svc.DeleteRoom(ctx, &livekit.DeleteRoomRequest{
 			Room: "testroom",
 		})

pkg/service/rtcservice.go

@@ -1,6 +1,7 @@
 package service
 
 import (
+	"context"
 	"fmt"
 	"io"
 	"net/http"
@@ -23,18 +24,28 @@ import (
 type RTCService struct {
 	router        routing.MessageRouter
 	roomAllocator RoomAllocator
+	store         RoomStore
 	upgrader      websocket.Upgrader
 	currentNode   routing.LocalNode
+	config        *config.Config
 	isDev         bool
 	limits        config.LimitConfig
 }
 
-func NewRTCService(conf *config.Config, ra RoomAllocator, router routing.MessageRouter, currentNode routing.LocalNode) *RTCService {
+func NewRTCService(
+	conf *config.Config,
+	ra RoomAllocator,
+	store RoomStore,
+	router routing.MessageRouter,
+	currentNode routing.LocalNode,
+) *RTCService {
 	s := &RTCService{
 		router:        router,
 		roomAllocator: ra,
+		store:         store,
 		upgrader:      websocket.Upgrader{},
 		currentNode:   currentNode,
+		config:        conf,
 		isDev:         conf.Development,
 		limits:        conf.Limit,
 	}
@@ -130,6 +141,16 @@ func (s *RTCService) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
+	// when autocreate is disabled, we'll check to ensure it's already created
+	if !s.config.Room.AutoCreate {
+		_, err := s.store.LoadRoom(context.Background(), roomName)
+		if err == ErrRoomNotFound {
+			handleError(w, 404, err.Error())
+		} else if err != nil {
+			handleError(w, 500, err.Error())
+		}
+	}
+
 	// create room if it doesn't exist, also assigns an RTC node for the room
 	rm, err := s.roomAllocator.CreateRoom(r.Context(), &livekit.CreateRoomRequest{Name: string(roomName)})
 	if err != nil {

pkg/service/wire_gen.go

@@ -50,7 +50,7 @@ func InitializeServer(conf *config.Config, currentNode routing.LocalNode) (*Live
 	analyticsService := telemetry.NewAnalyticsService(conf, currentNode)
 	telemetryService := telemetry.NewTelemetryService(notifier, analyticsService)
 	recordingService := NewRecordingService(messageBus, telemetryService)
-	rtcService := NewRTCService(conf, roomAllocator, router, currentNode)
+	rtcService := NewRTCService(conf, roomAllocator, roomStore, router, currentNode)
 	roomManager, err := NewLocalRoomManager(conf, roomStore, currentNode, router, telemetryService)
 	if err != nil {
 		return nil, err

test/integration_helpers.go

@@ -46,9 +46,9 @@ func init() {
 	serverlogger.InitFromConfig(config.LoggingConfig{Level: "debug"})
 }
 
-func setupSingleNodeTest(name string, roomName string) (*service.LivekitServer, func()) {
+func setupSingleNodeTest(name string) (*service.LivekitServer, func()) {
 	logger.Infow("----------------STARTING TEST----------------", "test", name)
-	s := createSingleNodeServer()
+	s := createSingleNodeServer(nil)
 	go func() {
 		if err := s.Start(); err != nil {
 			logger.Errorw("server returned error", err)
@@ -57,11 +57,6 @@ func setupSingleNodeTest(name string, roomName string) (*service.LivekitServer,
 
 	waitForServerToStart(s)
 
-	// create test room
-	_, err := roomClient.CreateRoom(contextWithToken(createRoomToken()), &livekit.CreateRoomRequest{Name: roomName})
-	if err != nil {
-		panic(err)
-	}
 	return s, func() {
 		s.Stop(true)
 		logger.Infow("----------------FINISHING TEST----------------", "test", name)
@@ -132,7 +127,7 @@ func waitUntilConnected(t *testing.T, clients ...*testclient.RTCClient) {
 	}
 }
 
-func createSingleNodeServer() *service.LivekitServer {
+func createSingleNodeServer(configUpdater func(*config.Config)) *service.LivekitServer {
 	var err error
 	conf, err := config.NewConfig("", nil)
 	if err != nil {
@@ -140,6 +135,9 @@ func createSingleNodeServer() *service.LivekitServer {
 	}
 	conf.Development = true
 	conf.Keys = map[string]string{testApiKey: testApiSecret}
+	if configUpdater != nil {
+		configUpdater(conf)
+	}
 
 	currentNode, err := routing.NewLocalNode(conf)
 	if err != nil {

test/multinode_test.go

@@ -151,3 +151,15 @@ func TestMultiNodeRoomList(t *testing.T) {
 
 	roomServiceListRoom(t)
 }
+
+func TestMultiNodeJoinAfterClose(t *testing.T) {
+	if testing.Short() {
+		t.SkipNow()
+		return
+	}
+
+	_, _, finish := setupMultiNodeTest("TestMultiNodeJoinAfterClose")
+	defer finish()
+
+	scenarioJoinClosedRoom(t)
+}

test/scenarios.go

@@ -4,13 +4,12 @@ import (
 	"testing"
 	"time"
 
+	"github.com/livekit/livekit-server/pkg/testutils"
+	testclient "github.com/livekit/livekit-server/test/client"
 	"github.com/livekit/protocol/livekit"
 	"github.com/livekit/protocol/logger"
 	"github.com/livekit/protocol/utils"
 	"github.com/stretchr/testify/require"
-
-	"github.com/livekit/livekit-server/pkg/testutils"
-	testclient "github.com/livekit/livekit-server/test/client"
 )
 
 // a scenario with lots of clients connecting, publishing, and leaving at random periods
@@ -141,6 +140,22 @@ func scenarioDataPublish(t *testing.T) {
 	})
 }
 
+func scenarioJoinClosedRoom(t *testing.T) {
+	c1 := createRTCClient("jcr1", defaultServerPort, nil)
+	waitUntilConnected(t, c1)
+
+	// close room with room client
+	_, err := roomClient.DeleteRoom(contextWithToken(createRoomToken()), &livekit.DeleteRoomRequest{
+		Room: testRoom,
+	})
+	require.NoError(t, err)
+
+	// now join again
+	c2 := createRTCClient("jcr2", defaultServerPort, nil)
+	waitUntilConnected(t, c2)
+	stopClients(c2)
+}
+
 func publishTracksForClients(t *testing.T, clients ...*testclient.RTCClient) []*testclient.TrackWriter {
 	logger.Infow("publishing tracks for clients")
 	var writers []*testclient.TrackWriter

test/singlenode_test.go

@@ -8,8 +8,10 @@ import (
 	"testing"
 	"time"
 
+	"github.com/livekit/livekit-server/pkg/config"
 	"github.com/livekit/protocol/auth"
 	"github.com/livekit/protocol/livekit"
+	"github.com/livekit/protocol/logger"
 	"github.com/pion/webrtc/v3"
 	"github.com/stretchr/testify/require"
 	"github.com/thoas/go-funk"
@@ -25,7 +27,7 @@ func TestClientCouldConnect(t *testing.T) {
 		return
 	}
 
-	_, finish := setupSingleNodeTest("TestClientCouldConnect", testRoom)
+	_, finish := setupSingleNodeTest("TestClientCouldConnect")
 	defer finish()
 
 	c1 := createRTCClient("c1", defaultServerPort, nil)
@@ -44,7 +46,7 @@ func TestClientConnectDuplicate(t *testing.T) {
 		return
 	}
 
-	_, finish := setupSingleNodeTest("TestClientCouldConnect", testRoom)
+	_, finish := setupSingleNodeTest("TestClientCouldConnect")
 	defer finish()
 
 	grant := &auth.VideoGrant{RoomJoin: true, Room: testRoom}
@@ -120,7 +122,7 @@ func TestSinglePublisher(t *testing.T) {
 		return
 	}
 
-	s, finish := setupSingleNodeTest("TestSinglePublisher", testRoom)
+	s, finish := setupSingleNodeTest("TestSinglePublisher")
 	defer finish()
 
 	c1 := createRTCClient("c1", defaultServerPort, nil)
@@ -203,7 +205,7 @@ func Test_WhenAutoSubscriptionDisabled_ClientShouldNotReceiveAnyPublishedTracks(
 		return
 	}
 
-	_, finish := setupSingleNodeTest("Test_WhenAutoSubscriptionDisabled_ClientShouldNotReceiveAnyPublishedTracks", testRoom)
+	_, finish := setupSingleNodeTest("Test_WhenAutoSubscriptionDisabled_ClientShouldNotReceiveAnyPublishedTracks")
 	defer finish()
 
 	opts := testclient.Options{AutoSubscribe: false}
@@ -228,7 +230,7 @@ func Test_RenegotiationWithDifferentCodecs(t *testing.T) {
 		return
 	}
 
-	_, finish := setupSingleNodeTest("TestRenegotiationWithDifferentCodecs", testRoom)
+	_, finish := setupSingleNodeTest("TestRenegotiationWithDifferentCodecs")
 	defer finish()
 
 	c1 := createRTCClient("c1", defaultServerPort, nil)
@@ -302,7 +304,7 @@ func TestSingleNodeRoomList(t *testing.T) {
 		t.SkipNow()
 		return
 	}
-	_, finish := setupSingleNodeTest("TestSingleNodeRoomList", testRoom)
+	_, finish := setupSingleNodeTest("TestSingleNodeRoomList")
 	defer finish()
 
 	roomServiceListRoom(t)
@@ -314,7 +316,7 @@ func TestSingleNodeCORS(t *testing.T) {
 		t.SkipNow()
 		return
 	}
-	s, finish := setupSingleNodeTest("TestSingleNodeCORS", testRoom)
+	s, finish := setupSingleNodeTest("TestSingleNodeCORS")
 	defer finish()
 
 	req, err := http.NewRequest("POST", fmt.Sprintf("http://localhost:%d", s.HTTPPort()), nil)
@@ -325,3 +327,57 @@ func TestSingleNodeCORS(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, "testhost.com", res.Header.Get("Access-Control-Allow-Origin"))
 }
+
+func TestSingleNodeJoinAfterClose(t *testing.T) {
+	if testing.Short() {
+		t.SkipNow()
+		return
+	}
+
+	_, finish := setupSingleNodeTest("TestJoinAfterClose")
+	defer finish()
+
+	scenarioJoinClosedRoom(t)
+}
+
+func TestAutoCreate(t *testing.T) {
+	disableAutoCreate := func(conf *config.Config) {
+		conf.Room.AutoCreate = false
+	}
+	t.Run("cannot join if room isn't created", func(t *testing.T) {
+		s := createSingleNodeServer(disableAutoCreate)
+		go func() {
+			if err := s.Start(); err != nil {
+				logger.Errorw("server returned error", err)
+			}
+		}()
+		defer s.Stop(true)
+
+		waitForServerToStart(s)
+
+		token := joinToken(testRoom, "start-before-create")
+		_, err := testclient.NewWebSocketConn(fmt.Sprintf("ws://localhost:%d", defaultServerPort), token, nil)
+		require.Error(t, err)
+	})
+
+	t.Run("join with explicit createRoom", func(t *testing.T) {
+		s := createSingleNodeServer(disableAutoCreate)
+		go func() {
+			if err := s.Start(); err != nil {
+				logger.Errorw("server returned error", err)
+			}
+		}()
+		defer s.Stop(true)
+
+		waitForServerToStart(s)
+
+		// explicitly create
+		_, err := roomClient.CreateRoom(contextWithToken(createRoomToken()), &livekit.CreateRoomRequest{Name: testRoom})
+		require.NoError(t, err)
+
+		c1 := createRTCClient("join-after-create", defaultServerPort, nil)
+		waitUntilConnected(t, c1)
+
+		c1.Stop()
+	})
+}