close rooms after delay instead of immediately after participant departure, to allow reconnects

pkg/rtc/participant_internal_test.go

@@ -12,6 +12,8 @@ import (
 	"github.com/livekit/livekit-server/proto/livekit"
 )
 
+//func Test
+
 func TestIsReady(t *testing.T) {
 	tests := []struct {
 		state livekit.ParticipantInfo_State

pkg/rtc/room.go

@@ -2,6 +2,7 @@ package rtc
 
 import (
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/thoas/go-funk"
@@ -12,24 +13,38 @@ import (
 	"github.com/livekit/livekit-server/proto/livekit"
 )
 
+const (
+	DefaultEmptyTimeout = 5 * 60 // 5 mins
+)
+
 type Room struct {
 	livekit.Room
 	config WebRTCConfig
 	lock   sync.RWMutex
 	// map of identity -> Participant
 	participants map[string]types.Participant
-	hasJoined    bool
-	isClosed     utils.AtomicFlag
-	onClose      func()
+	// time the first participant joined the room
+	joinedAt atomic.Value
+	// time that the last participant left the room
+	leftAt   atomic.Value
+	isClosed utils.AtomicFlag
+	onClose  func()
 }
 
 func NewRoom(room *livekit.Room, config WebRTCConfig) *Room {
-	return &Room{
+	r := &Room{
 		Room:         *room,
 		config:       config,
 		lock:         sync.RWMutex{},
 		participants: make(map[string]types.Participant),
 	}
+	if r.EmptyTimeout == 0 {
+		r.EmptyTimeout = DefaultEmptyTimeout
+	}
+	if r.CreationTime == 0 {
+		r.CreationTime = time.Now().Unix()
+	}
+	return r
 }
 
 func (r *Room) GetParticipant(identity string) types.Participant {
@@ -44,6 +59,22 @@ func (r *Room) GetParticipants() []types.Participant {
 	return funk.Values(r.participants).([]types.Participant)
 }
 
+func (r *Room) FirstJoinedAt() int64 {
+	j := r.joinedAt.Load()
+	if t, ok := j.(int64); ok {
+		return t
+	}
+	return 0
+}
+
+func (r *Room) LastLeftAt() int64 {
+	l := r.leftAt.Load()
+	if t, ok := l.(int64); ok {
+		return t
+	}
+	return 0
+}
+
 func (r *Room) Join(participant types.Participant) error {
 	if r.isClosed.Get() {
 		return ErrRoomClosed
@@ -54,14 +85,15 @@ func (r *Room) Join(participant types.Participant) error {
 
 	if r.participants[participant.Identity()] != nil {
 		return ErrAlreadyJoined
-
 	}
 
 	if r.MaxParticipants > 0 && int(r.MaxParticipants) == len(r.participants) {
 		return ErrMaxParticipantsExceeded
 	}
 
-	r.hasJoined = true
+	if r.FirstJoinedAt() == 0 {
+		r.joinedAt.Store(time.Now().Unix())
+	}
 
 	// it's important to set this before connection, we don't want to miss out on any publishedTracks
 	participant.OnTrackPublished(r.onTrackAdded)
@@ -126,7 +158,9 @@ func (r *Room) RemoveParticipant(identity string) {
 
 	delete(r.participants, identity)
 
-	go r.CloseIfEmpty()
+	if len(r.participants) == 0 {
+		r.leftAt.Store(time.Now().Unix())
+	}
 }
 
 // Close the room if all participants had left, or it's still empty past timeout
@@ -143,8 +177,14 @@ func (r *Room) CloseIfEmpty() {
 		return
 	}
 
-	elapsed := uint32(time.Now().Unix() - r.CreationTime)
-	if r.hasJoined || (r.EmptyTimeout > 0 && elapsed >= r.EmptyTimeout) {
+	var elapsed int64
+	if r.FirstJoinedAt() > 0 {
+		// compute elasped from last departure
+		elapsed = time.Now().Unix() - r.LastLeftAt()
+	} else {
+		elapsed = time.Now().Unix() - r.CreationTime
+	}
+	if elapsed >= int64(r.EmptyTimeout) {
 		r.Close()
 	}
 }

pkg/rtc/room_test.go

@@ -7,6 +7,7 @@ import (
 
 	"github.com/stretchr/testify/assert"
 
+	"github.com/livekit/livekit-server/pkg/logger"
 	"github.com/livekit/livekit-server/pkg/rtc"
 	"github.com/livekit/livekit-server/pkg/rtc/types/typesfakes"
 	"github.com/livekit/livekit-server/proto/livekit"
@@ -17,15 +18,49 @@ const (
 	defaultDelay    = 10 * time.Millisecond
 )
 
+func init() {
+	logger.InitDevelopment("")
+}
+
+func TestJoinedState(t *testing.T) {
+	t.Run("new room should return joinedAt 0", func(t *testing.T) {
+		rm := newRoomWithParticipants(t, 0)
+		assert.Equal(t, int64(0), rm.FirstJoinedAt())
+		assert.Equal(t, int64(0), rm.LastLeftAt())
+	})
+
+	t.Run("should be current time when a participant joins", func(t *testing.T) {
+		s := time.Now().Unix()
+		rm := newRoomWithParticipants(t, 1)
+		assert.Equal(t, s, rm.FirstJoinedAt())
+		assert.Equal(t, int64(0), rm.LastLeftAt())
+	})
+
+	t.Run("should be set when a participant leaves", func(t *testing.T) {
+		rm := newRoomWithParticipants(t, 1)
+		p0 := rm.GetParticipants()[0]
+		s := time.Now().Unix()
+		rm.RemoveParticipant(p0.Identity())
+		assert.Equal(t, s, rm.LastLeftAt())
+	})
+
+	t.Run("LastLeftAt should not be set when there are still participants in the room", func(t *testing.T) {
+		rm := newRoomWithParticipants(t, 2)
+		p0 := rm.GetParticipants()[0]
+		rm.RemoveParticipant(p0.Identity())
+		assert.EqualValues(t, 0, rm.LastLeftAt())
+	})
+}
+
 func TestRoomJoin(t *testing.T) {
 	t.Run("joining returns existing participant data", func(t *testing.T) {
 		rm := newRoomWithParticipants(t, numParticipants)
-		p := newMockParticipant("new")
+		pNew := newMockParticipant("new")
 
-		rm.Join(p)
+		rm.Join(pNew)
 
 		// expect new participant to get a JoinReply
-		info, participants := p.SendJoinResponseArgsForCall(0)
+		info, participants := pNew.SendJoinResponseArgsForCall(0)
 		assert.Equal(t, info.Sid, rm.Sid)
 		assert.Len(t, participants, numParticipants)
 		assert.Len(t, rm.GetParticipants(), numParticipants+1)
@@ -98,10 +133,13 @@ func TestRoomClosure(t *testing.T) {
 			isClosed = true
 		})
 		p := rm.GetParticipants()[0]
+		// allows immediate close after
+		rm.EmptyTimeout = 0
 		rm.RemoveParticipant(p.Identity())
 
 		time.Sleep(defaultDelay)
 
+		rm.CloseIfEmpty()
 		assert.Len(t, rm.GetParticipants(), 0)
 		assert.True(t, isClosed)
 
@@ -114,7 +152,7 @@ func TestRoomClosure(t *testing.T) {
 		rm.OnClose(func() {
 			isClosed = true
 		})
-
+		assert.NotZero(t, rm.EmptyTimeout)
 		rm.CloseIfEmpty()
 		assert.False(t, isClosed)
 	})
@@ -156,7 +194,6 @@ func TestNewTrack(t *testing.T) {
 }
 
 func newRoomWithParticipants(t *testing.T, num int) *rtc.Room {
-
 	rm := rtc.NewRoom(
 		&livekit.Room{Name: "room"},
 		rtc.WebRTCConfig{},

pkg/service/roommanager.go

@@ -4,6 +4,8 @@ import (
 	"sync"
 	"time"
 
+	"github.com/thoas/go-funk"
+
 	"github.com/livekit/livekit-server/pkg/logger"
 	"github.com/livekit/livekit-server/pkg/routing"
 	"github.com/livekit/livekit-server/pkg/rtc"
@@ -98,6 +100,7 @@ func (r *RoomManager) DeleteRoom(roomName string) error {
 		err2 = r.roomStore.DeleteRoom(roomName)
 	}()
 
+	wg.Wait()
 	if err != nil {
 		err = err2
 	}
@@ -106,7 +109,7 @@ func (r *RoomManager) DeleteRoom(roomName string) error {
 }
 
 // clean up after old rooms that have been around for awhile
-func (r *RoomManager) Cleanup() error {
+func (r *RoomManager) CleanupRooms() error {
 	// cleanup rooms that have been left for over a day
 	rooms, err := r.roomStore.ListRooms()
 	if err != nil {
@@ -124,6 +127,16 @@ func (r *RoomManager) Cleanup() error {
 	return nil
 }
 
+func (r *RoomManager) CloseIdleRooms() {
+	r.lock.RLock()
+	rooms := funk.Values(r.rooms).([]*rtc.Room)
+	r.lock.RUnlock()
+
+	for _, room := range rooms {
+		room.CloseIfEmpty()
+	}
+}
+
 // starts WebRTC session when a new participant is connected, takes place on RTC node
 func (r *RoomManager) StartSession(roomName, identity string, reconnect bool, requestSource routing.MessageSource, responseSink routing.MessageSink) {
 	room, err := r.getOrCreateRoom(roomName)

pkg/service/server.go

@@ -14,6 +14,7 @@ import (
 	"github.com/livekit/livekit-server/pkg/config"
 	"github.com/livekit/livekit-server/pkg/logger"
 	"github.com/livekit/livekit-server/pkg/routing"
+	"github.com/livekit/livekit-server/pkg/utils"
 	"github.com/livekit/livekit-server/proto/livekit"
 	"github.com/livekit/livekit-server/version"
 )
@@ -24,8 +25,9 @@ type LivekitServer struct {
 	rtcService  *RTCService
 	httpServer  *http.Server
 	router      routing.Router
+	roomManager *RoomManager
 	currentNode routing.LocalNode
-	running     bool
+	running     utils.AtomicFlag
 	doneChan    chan bool
 }
 
@@ -42,6 +44,7 @@ func NewLivekitServer(conf *config.Config,
 		roomServer:  livekit.NewRoomServiceServer(roomService),
 		rtcService:  rtcService,
 		router:      router,
+		roomManager: roomManager,
 		currentNode: currentNode,
 	}
 
@@ -66,17 +69,17 @@ func NewLivekitServer(conf *config.Config,
 	router.OnNewParticipantRTC(roomManager.StartSession)
 
 	// clean up old rooms on startup
-	err = roomManager.Cleanup()
+	err = roomManager.CleanupRooms()
 
 	return
 }
 
 func (s *LivekitServer) IsRunning() bool {
-	return s.running
+	return s.running.Get()
 }
 
 func (s *LivekitServer) Start() error {
-	if s.running {
+	if s.running.Get() {
 		return errors.New("already running")
 	}
 
@@ -104,7 +107,9 @@ func (s *LivekitServer) Start() error {
 		s.httpServer.Serve(ln)
 	}()
 
-	s.running = true
+	go s.backgroundWorker()
+
+	s.running.TrySet(true)
 
 	<-s.doneChan
 
@@ -120,9 +125,23 @@ func (s *LivekitServer) Start() error {
 }
 
 func (s *LivekitServer) Stop() {
-	s.running = false
-	s.router.Stop()
-	s.doneChan <- true
+	if s.running.TrySet(false) {
+		s.router.Stop()
+		close(s.doneChan)
+	}
+}
+
+// worker to perform periodic tasks per node
+func (s *LivekitServer) backgroundWorker() {
+	roomTicker := time.NewTicker(30 * time.Second)
+	for {
+		select {
+		case <-s.doneChan:
+			return
+		case <-roomTicker.C:
+			s.roomManager.CloseIdleRooms()
+		}
+	}
 }
 
 func configureMiddlewares(handler http.Handler, middlewares ...negroni.Handler) *negroni.Negroni {

test/multinode_test.go

@@ -1,14 +1,10 @@
 package test
 
 import (
-	"context"
 	"testing"
 
-	"github.com/go-redis/redis/v8"
 	"github.com/stretchr/testify/assert"
 
-	"github.com/livekit/livekit-server/pkg/routing"
-	"github.com/livekit/livekit-server/pkg/service"
 	"github.com/livekit/livekit-server/proto/livekit"
 )
 
@@ -58,18 +54,21 @@ func TestMultiNodeRouting(t *testing.T) {
 	c1.Stop()
 	c2.Stop()
 
-	// ensure that room is closed
-	rc := redisClient()
-	ctx := context.Background()
-	withTimeout(t, "room should be closed", func() bool {
-		if rc.HGet(ctx, service.RoomsKey, testRoom).Err() == nil {
-			return false
-		}
-		return true
-	})
-
-	assert.Equal(t, redis.Nil, rc.HGet(ctx, routing.NodeRoomKey, testRoom).Err())
-	assert.Equal(t, redis.Nil, rc.HGet(ctx, service.RoomIdMap, testRoom).Err())
+	// TODO: delete room explicitly and ensure it's closed
+	//
+	//// ensure that room is closed
+	//
+	//rc := redisClient()
+	//ctx := context.Background()
+	//withTimeout(t, "room should be closed", func() bool {
+	//	if rc.HGet(ctx, service.RoomsKey, testRoom).Err() == nil {
+	//		return false
+	//	}
+	//	return true
+	//})
+	//
+	//assert.Equal(t, redis.Nil, rc.HGet(ctx, routing.NodeRoomKey, testRoom).Err())
+	//assert.Equal(t, redis.Nil, rc.HGet(ctx, service.RoomIdMap, testRoom).Err())
 }
 
 func TestConnectWithoutCreation(t *testing.T) {