telemetry: support roomID change for a participant (#4816)

* telemetry: support roomID change for a participant

A room can get a new id while participants are connected. Key stats
workers as map[roomID]map[participantID] so moving a room is a single
map splice, and add reKeyRoom/RoomIDChanged to do the move.

Stats collected before the change are sealed off with the room they
were collected in so they stay attributed to it.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* telemetry: close superseded worker on re-key collision

Only one worker can be keyed at (room, participant). If a re-key lands
on a room that already has a worker for the same participant, keep the
one already filed there and close the superseded one so it drains and
is reaped instead of lingering in the flush list.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* telemetry: hand references to the successor on force close

A ReferenceGuard records that it activated some worker, not which one,
so a superseded worker cannot just drop its references - the survivor
would be left with references it never sees released and would never
close. Hand them over instead.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>

pkg/telemetry/events.go

@@ -184,6 +184,18 @@ func (t *telemetryService) ParticipantResumed(
 	})
 }
 
+// RoomIDChanged re-keys the room's stats workers.
+//
+// NOTE: this shares the queue with the stats and participant events it races with, so
+// ops raised before the id changed (carrying `prevRoomID`) are applied before the
+// re-key and ops raised after it (carrying the new id) are applied after. Callers
+// should raise this as soon as the room starts reporting the new id.
+func (t *telemetryService) RoomIDChanged(ctx context.Context, prevRoomID livekit.RoomID, room *livekit.Room) {
+	t.enqueue(func() {
+		t.reKeyRoom(prevRoomID, livekit.RoomID(room.Sid), livekit.RoomName(room.Name))
+	})
+}
+
 func (t *telemetryService) ParticipantLeft(ctx context.Context,
 	room *livekit.Room,
 	participant *livekit.ParticipantInfo,

pkg/telemetry/stats_test.go

@@ -626,6 +626,133 @@ func Test_BothDownstreamAndUpstreamStatsAreSentTogether(t *testing.T) {
 	require.Equal(t, livekit.StreamType_DOWNSTREAM, stats[1].Kind)
 }
 
+func Test_RoomIDChangeReKeysStatsWorkers(t *testing.T) {
+	fixture := createFixture()
+
+	// prepare
+	room := &livekit.Room{Sid: "RoomSid", Name: "RoomName"}
+	partSID := livekit.ParticipantID("part1")
+	participantInfo := &livekit.ParticipantInfo{Sid: string(partSID)}
+	trackID := livekit.TrackID("trackID")
+	guard := &telemetry.ReferenceGuard{}
+	fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil, nil, true, guard)
+
+	stat1 := &livekit.AnalyticsStat{Streams: []*livekit.AnalyticsStream{{PrimaryBytes: 33}}}
+	fixture.sut.TrackStats(livekit.RoomID(room.Sid), livekit.RoomName(room.Name), telemetry.StatsKeyForData("test", livekit.StreamType_DOWNSTREAM, partSID, trackID), stat1)
+
+	// do - the room restarts and gets a new id
+	restartedRoom := &livekit.Room{Sid: "RestartedSid", Name: "RoomName"}
+	fixture.sut.RoomIDChanged(context.Background(), livekit.RoomID(room.Sid), restartedRoom)
+
+	// stats reported with the new id reach the same worker
+	stat2 := &livekit.AnalyticsStat{Streams: []*livekit.AnalyticsStream{{PrimaryBytes: 44}}}
+	fixture.sut.TrackStats(livekit.RoomID(restartedRoom.Sid), livekit.RoomName(restartedRoom.Name), telemetry.StatsKeyForData("test", livekit.StreamType_DOWNSTREAM, partSID, trackID), stat2)
+
+	fixture.flush()
+
+	// one worker, one flush, but two stats - each attributed to the session it was collected in
+	require.Equal(t, 1, fixture.analytics.SendStatsCallCount())
+	_, stats := fixture.analytics.SendStatsArgsForCall(0)
+	require.Equal(t, 2, len(stats))
+
+	byRoom := map[string]*livekit.AnalyticsStat{}
+	for _, stat := range stats {
+		require.Equal(t, string(partSID), stat.ParticipantId)
+		byRoom[stat.RoomId] = stat
+	}
+	require.Len(t, byRoom, 2)
+	require.Equal(t, uint64(33), byRoom[room.Sid].Streams[0].PrimaryBytes)
+	require.Equal(t, uint64(44), byRoom[restartedRoom.Sid].Streams[0].PrimaryBytes)
+
+	// the worker moved rather than being duplicated, so closing it out drains everything
+	fixture.sut.ParticipantLeft(context.Background(), restartedRoom, participantInfo, true, guard)
+	fixture.flush()
+	require.Equal(t, 1, fixture.analytics.SendStatsCallCount())
+}
+
+// a forwarded participant is in more than one room at a time under the same participant
+// id, so re-keying one of those rooms must leave the other alone
+func Test_RoomIDChangeLeavesForwardedParticipantAlone(t *testing.T) {
+	fixture := createFixture()
+
+	// prepare - the same participant id in a source room and a forwarding destination room
+	sourceRoom := &livekit.Room{Sid: "SourceSid", Name: "SourceRoom"}
+	destRoom := &livekit.Room{Sid: "DestSid", Name: "DestRoom"}
+	partSID := livekit.ParticipantID("part1")
+	participantInfo := &livekit.ParticipantInfo{Sid: string(partSID)}
+	trackID := livekit.TrackID("trackID")
+	fixture.sut.ParticipantJoined(context.Background(), sourceRoom, participantInfo, nil, nil, true, &telemetry.ReferenceGuard{})
+	fixture.sut.ParticipantJoined(context.Background(), destRoom, participantInfo, nil, nil, true, &telemetry.ReferenceGuard{})
+
+	// do - only the destination room restarts
+	restartedDest := &livekit.Room{Sid: "RestartedDestSid", Name: "DestRoom"}
+	fixture.sut.RoomIDChanged(context.Background(), livekit.RoomID(destRoom.Sid), restartedDest)
+
+	stat1 := &livekit.AnalyticsStat{Streams: []*livekit.AnalyticsStream{{PrimaryBytes: 33}}}
+	fixture.sut.TrackStats(livekit.RoomID(sourceRoom.Sid), livekit.RoomName(sourceRoom.Name), telemetry.StatsKeyForData("test", livekit.StreamType_DOWNSTREAM, partSID, trackID), stat1)
+	stat2 := &livekit.AnalyticsStat{Streams: []*livekit.AnalyticsStream{{PrimaryBytes: 44}}}
+	fixture.sut.TrackStats(livekit.RoomID(restartedDest.Sid), livekit.RoomName(restartedDest.Name), telemetry.StatsKeyForData("test", livekit.StreamType_DOWNSTREAM, partSID, trackID), stat2)
+
+	fixture.flush()
+
+	// the source room's worker is untouched, the destination room's worker moved
+	byRoom := map[string]*livekit.AnalyticsStat{}
+	for i := 0; i < fixture.analytics.SendStatsCallCount(); i++ {
+		_, stats := fixture.analytics.SendStatsArgsForCall(i)
+		for _, stat := range stats {
+			byRoom[stat.RoomId] = stat
+		}
+	}
+	require.Len(t, byRoom, 2)
+	require.Equal(t, uint64(33), byRoom[sourceRoom.Sid].Streams[0].PrimaryBytes)
+	require.Equal(t, sourceRoom.Name, byRoom[sourceRoom.Sid].RoomName)
+	require.Equal(t, uint64(44), byRoom[restartedDest.Sid].Streams[0].PrimaryBytes)
+	require.Equal(t, restartedDest.Name, byRoom[restartedDest.Sid].RoomName)
+}
+
+// a re-key should never land on a room that already has workers, but if it does only
+// one worker can be keyed at (room, participant) and the superseded one must not be
+// left unreachable in the flush list
+func Test_RoomIDChangeParticipantCollision(t *testing.T) {
+	fixture := createFixture()
+
+	// prepare - the same participant id in the room being re-keyed and in its destination
+	prevRoom := &livekit.Room{Sid: "PrevSid", Name: "PrevRoom"}
+	destRoom := &livekit.Room{Sid: "DestSid", Name: "DestRoom"}
+	partSID := livekit.ParticipantID("part1")
+	participantInfo := &livekit.ParticipantInfo{Sid: string(partSID)}
+	trackID := livekit.TrackID("trackID")
+	fixture.sut.ParticipantJoined(context.Background(), prevRoom, participantInfo, nil, nil, true, &telemetry.ReferenceGuard{})
+	fixture.sut.ParticipantJoined(context.Background(), destRoom, participantInfo, nil, nil, true, &telemetry.ReferenceGuard{})
+
+	stat1 := &livekit.AnalyticsStat{Streams: []*livekit.AnalyticsStream{{PrimaryBytes: 33}}}
+	fixture.sut.TrackStats(livekit.RoomID(prevRoom.Sid), livekit.RoomName(prevRoom.Name), telemetry.StatsKeyForData("test", livekit.StreamType_DOWNSTREAM, partSID, trackID), stat1)
+
+	// do
+	fixture.sut.RoomIDChanged(context.Background(), livekit.RoomID(prevRoom.Sid), destRoom)
+
+	// the superseded worker drains what it collected under the room it was in
+	fixture.flush()
+	byRoom := map[string]*livekit.AnalyticsStat{}
+	for i := 0; i < fixture.analytics.SendStatsCallCount(); i++ {
+		_, stats := fixture.analytics.SendStatsArgsForCall(i)
+		for _, stat := range stats {
+			byRoom[stat.RoomId] = stat
+		}
+	}
+	require.Equal(t, uint64(33), byRoom[prevRoom.Sid].Streams[0].PrimaryBytes)
+
+	// the worker already keyed at the destination wins and keeps receiving stats
+	stat2 := &livekit.AnalyticsStat{Streams: []*livekit.AnalyticsStream{{PrimaryBytes: 44}}}
+	fixture.sut.TrackStats(livekit.RoomID(destRoom.Sid), livekit.RoomName(destRoom.Name), telemetry.StatsKeyForData("test", livekit.StreamType_DOWNSTREAM, partSID, trackID), stat2)
+
+	fixture.flush()
+	_, stats := fixture.analytics.SendStatsArgsForCall(fixture.analytics.SendStatsCallCount() - 1)
+	require.Equal(t, 1, len(stats))
+	require.Equal(t, destRoom.Sid, stats[0].RoomId)
+	require.Equal(t, uint64(44), stats[0].Streams[0].PrimaryBytes)
+}
+
 func (f *telemetryServiceFixture) flush() {
 	time.Sleep(time.Millisecond * 500)
 	f.sut.FlushStats()

pkg/telemetry/statsworker.go

@@ -52,6 +52,18 @@ func (s *ReferenceCount) Release(guard *ReferenceGuard) bool {
 	return s.count == 0
 }
 
+// Take hands over every reference held, leaving none behind.
+func (s *ReferenceCount) Take() int {
+	count := s.count
+	s.count = 0
+	return count
+}
+
+// Absorb takes on references handed over from elsewhere.
+func (s *ReferenceCount) Absorb(count int) {
+	s.count += count
+}
+
 func (s ReferenceCount) MarshalLogObject(e zapcore.ObjectEncoder) error {
 	e.AddInt("count", s.count)
 	return nil
@@ -59,19 +71,39 @@ func (s ReferenceCount) MarshalLogObject(e zapcore.ObjectEncoder) error {
 
 // ----------------------------------------
 
+// statsBatch is stats collected while the worker was in one room
+type statsBatch struct {
+	roomID           livekit.RoomID
+	roomName         livekit.RoomName
+	incomingPerTrack map[livekit.TrackID][]*livekit.AnalyticsStat
+	outgoingPerTrack map[livekit.TrackID][]*livekit.AnalyticsStat
+}
+
+func (b statsBatch) isEmpty() bool {
+	return len(b.incomingPerTrack) == 0 && len(b.outgoingPerTrack) == 0
+}
+
+// ----------------------------------------
+
 // StatsWorker handles participant stats
 type StatsWorker struct {
 	next *StatsWorker
 
 	ctx                 context.Context
 	t                   TelemetryService
-	roomID              livekit.RoomID
-	roomName            livekit.RoomName
 	participantID       livekit.ParticipantID
 	participantIdentity livekit.ParticipantIdentity
-	isConnected         bool
 
-	lock             sync.RWMutex
+	lock sync.RWMutex
+	// the room a worker belongs to can change mid-session, so it is mutable state
+	// guarded by `lock`. it is kept in sync with the key the worker is filed under in
+	// telemetryService.workers, see telemetryService.reKeyRoom.
+	roomID   livekit.RoomID
+	roomName livekit.RoomName
+	// batches sealed off by a room change, they carry the room they were collected
+	// in and go out on the next flush
+	sealed           []statsBatch
+	isConnected      bool
 	outgoingPerTrack map[livekit.TrackID][]*livekit.AnalyticsStat
 	incomingPerTrack map[livekit.TrackID][]*livekit.AnalyticsStat
 	refCount         ReferenceCount
@@ -115,6 +147,51 @@ func (s *StatsWorker) ParticipantID() livekit.ParticipantID {
 	return s.participantID
 }
 
+func (s *StatsWorker) RoomID() livekit.RoomID {
+	s.lock.RLock()
+	defer s.lock.RUnlock()
+
+	return s.roomID
+}
+
+// SetRoom re-points the worker at a room.
+//
+// Stats collected so far are sealed off rather than re-stamped - a room id changes
+// because the previous session ended, so what was collected under it belongs to it.
+// Sealing keeps the re-key free of any sending, the sealed stats go out on the next
+// flush like every other stat.
+func (s *StatsWorker) SetRoom(roomID livekit.RoomID, roomName livekit.RoomName) {
+	s.lock.Lock()
+	defer s.lock.Unlock()
+
+	if s.roomID == roomID && s.roomName == roomName {
+		return
+	}
+
+	if batch := s.sealStatsLocked(); !batch.isEmpty() {
+		s.sealed = append(s.sealed, batch)
+	}
+
+	s.roomID = roomID
+	s.roomName = roomName
+}
+
+// sealStatsLocked hands over everything collected since the last seal, stamped
+// with the room it was collected in
+func (s *StatsWorker) sealStatsLocked() statsBatch {
+	batch := statsBatch{
+		roomID:           s.roomID,
+		roomName:         s.roomName,
+		incomingPerTrack: s.incomingPerTrack,
+		outgoingPerTrack: s.outgoingPerTrack,
+	}
+
+	s.incomingPerTrack = make(map[livekit.TrackID][]*livekit.AnalyticsStat)
+	s.outgoingPerTrack = make(map[livekit.TrackID][]*livekit.AnalyticsStat)
+
+	return batch
+}
+
 func (s *StatsWorker) SetConnected() {
 	s.lock.Lock()
 	s.isConnected = true
@@ -132,19 +209,24 @@ func (s *StatsWorker) Flush(now time.Time, closeWait time.Duration) bool {
 	ts := timestamppb.New(now)
 
 	s.lock.Lock()
-	stats := make([]*livekit.AnalyticsStat, 0, len(s.incomingPerTrack)+len(s.outgoingPerTrack))
-
-	incomingPerTrack := s.incomingPerTrack
-	s.incomingPerTrack = make(map[livekit.TrackID][]*livekit.AnalyticsStat)
-
-	outgoingPerTrack := s.outgoingPerTrack
-	s.outgoingPerTrack = make(map[livekit.TrackID][]*livekit.AnalyticsStat)
+	// anything sealed off by a room change goes out along with the current batch,
+	// each stamped with the room it was collected in
+	batches := append(s.sealed, s.sealStatsLocked())
+	s.sealed = nil
 
 	closed := !s.closedAt.IsZero() && now.Sub(s.closedAt) > closeWait
 	s.lock.Unlock()
 
-	stats = s.collectStats(ts, livekit.StreamType_UPSTREAM, incomingPerTrack, stats)
-	stats = s.collectStats(ts, livekit.StreamType_DOWNSTREAM, outgoingPerTrack, stats)
+	numTracks := 0
+	for _, batch := range batches {
+		numTracks += len(batch.incomingPerTrack) + len(batch.outgoingPerTrack)
+	}
+
+	stats := make([]*livekit.AnalyticsStat, 0, numTracks)
+	for _, batch := range batches {
+		stats = s.collectStats(ts, batch, livekit.StreamType_UPSTREAM, stats)
+		stats = s.collectStats(ts, batch, livekit.StreamType_DOWNSTREAM, stats)
+	}
 	if len(stats) > 0 {
 		s.t.SendStats(s.ctx, stats)
 	}
@@ -167,6 +249,34 @@ func (s *StatsWorker) Close(guard *ReferenceGuard) bool {
 	return ok
 }
 
+// ForceClose closes the worker irrespective of outstanding references. Used when a
+// worker can no longer be reached through the worker map, so that it drains and is
+// reaped instead of lingering in the flush list forever.
+//
+// Its references are handed over to `successor`, the worker that can be reached in its
+// place, so that whoever holds one still has a live worker to close. A ReferenceGuard
+// records that it activated some worker, not which one, so leaving them behind would
+// strand the successor with references it can never see released.
+func (s *StatsWorker) ForceClose(successor *StatsWorker) bool {
+	s.lock.Lock()
+	if !s.closedAt.IsZero() {
+		s.lock.Unlock()
+		return false
+	}
+
+	s.closedAt = time.Now()
+	count := s.refCount.Take()
+	s.lock.Unlock()
+
+	if successor != nil && count != 0 {
+		successor.lock.Lock()
+		successor.refCount.Absorb(count)
+		successor.lock.Unlock()
+	}
+
+	return true
+}
+
 func (s *StatsWorker) Closed(guard *ReferenceGuard) bool {
 	s.lock.Lock()
 	defer s.lock.Unlock()
@@ -179,10 +289,15 @@ func (s *StatsWorker) Closed(guard *ReferenceGuard) bool {
 
 func (s *StatsWorker) collectStats(
 	ts *timestamppb.Timestamp,
+	batch statsBatch,
 	streamType livekit.StreamType,
-	perTrack map[livekit.TrackID][]*livekit.AnalyticsStat,
 	stats []*livekit.AnalyticsStat,
 ) []*livekit.AnalyticsStat {
+	perTrack := batch.incomingPerTrack
+	if streamType == livekit.StreamType_DOWNSTREAM {
+		perTrack = batch.outgoingPerTrack
+	}
+
 	for trackID, analyticsStats := range perTrack {
 		coalesced := coalesce(analyticsStats)
 		if coalesced == nil {
@@ -192,9 +307,9 @@ func (s *StatsWorker) collectStats(
 		coalesced.TimeStamp = ts
 		coalesced.TrackId = string(trackID)
 		coalesced.Kind = streamType
-		coalesced.RoomId = string(s.roomID)
+		coalesced.RoomId = string(batch.roomID)
 		coalesced.ParticipantId = string(s.participantID)
-		coalesced.RoomName = string(s.roomName)
+		coalesced.RoomName = string(batch.roomName)
 		stats = append(stats, coalesced)
 	}
 	return stats

pkg/telemetry/statsworker_test.go

@@ -16,4 +16,48 @@ func TestStatsWorker(t *testing.T) {
 		require.True(t, w.Close(&g1))
 		require.True(t, w.Closed(&g1))
 	})
+
+	// a ReferenceGuard records that it activated some worker, not which one, so a
+	// superseded worker has to hand its references to the one reachable in its place
+	t.Run("force close hands references to the successor", func(t *testing.T) {
+		t.Run("a guard shared by both workers", func(t *testing.T) {
+			// the second worker never got a reference, the guard was already activated
+			var g ReferenceGuard
+			superseded := newStatsWorker(t.Context(), nil, "", "", "", "", &g)
+			survivor := newStatsWorker(t.Context(), nil, "", "", "", "", &g)
+			require.Equal(t, 1, superseded.refCount.count)
+			require.Equal(t, 0, survivor.refCount.count)
+
+			require.True(t, superseded.ForceClose(survivor))
+			require.Equal(t, 0, superseded.refCount.count)
+			require.Equal(t, 1, survivor.refCount.count)
+
+			// without the hand over this would leave the survivor at -1 and never closed
+			require.True(t, survivor.Close(&g))
+			require.True(t, survivor.Closed(&g))
+		})
+
+		t.Run("a guard per worker", func(t *testing.T) {
+			var gSuperseded, gSurvivor ReferenceGuard
+			superseded := newStatsWorker(t.Context(), nil, "", "", "", "", &gSuperseded)
+			survivor := newStatsWorker(t.Context(), nil, "", "", "", "", &gSurvivor)
+
+			require.True(t, superseded.ForceClose(survivor))
+			require.Equal(t, 2, survivor.refCount.count)
+
+			// the superseded worker's owner departs, it must not close the survivor early
+			require.False(t, survivor.Close(&gSuperseded))
+			require.True(t, survivor.Close(&gSurvivor))
+		})
+
+		t.Run("closing an already closed worker holds on to its references", func(t *testing.T) {
+			var g ReferenceGuard
+			superseded := newStatsWorker(t.Context(), nil, "", "", "", "", &g)
+			survivor := newStatsWorker(t.Context(), nil, "", "", "", "", nil)
+
+			require.True(t, superseded.ForceClose(nil))
+			require.False(t, superseded.ForceClose(survivor))
+			require.Equal(t, 0, survivor.refCount.count)
+		})
+	})
 }

pkg/telemetry/telemetryfakes/fake_telemetry_service.go

@@ -136,6 +136,13 @@ type FakeTelemetryService struct {
 		arg2 *livekit.Room
 		arg3 livekit.RoomEndReason
 	}
+	RoomIDChangedStub        func(context.Context, livekit.RoomID, *livekit.Room)
+	roomIDChangedMutex       sync.RWMutex
+	roomIDChangedArgsForCall []struct {
+		arg1 context.Context
+		arg2 livekit.RoomID
+		arg3 *livekit.Room
+	}
 	RoomProjectReporterStub        func(context.Context) roomobs.ProjectReporter
 	roomProjectReporterMutex       sync.RWMutex
 	roomProjectReporterArgsForCall []struct {
@@ -915,6 +922,40 @@ func (fake *FakeTelemetryService) RoomEndedArgsForCall(i int) (context.Context,
 	return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3
 }
 
+func (fake *FakeTelemetryService) RoomIDChanged(arg1 context.Context, arg2 livekit.RoomID, arg3 *livekit.Room) {
+	fake.roomIDChangedMutex.Lock()
+	fake.roomIDChangedArgsForCall = append(fake.roomIDChangedArgsForCall, struct {
+		arg1 context.Context
+		arg2 livekit.RoomID
+		arg3 *livekit.Room
+	}{arg1, arg2, arg3})
+	stub := fake.RoomIDChangedStub
+	fake.recordInvocation("RoomIDChanged", []interface{}{arg1, arg2, arg3})
+	fake.roomIDChangedMutex.Unlock()
+	if stub != nil {
+		fake.RoomIDChangedStub(arg1, arg2, arg3)
+	}
+}
+
+func (fake *FakeTelemetryService) RoomIDChangedCallCount() int {
+	fake.roomIDChangedMutex.RLock()
+	defer fake.roomIDChangedMutex.RUnlock()
+	return len(fake.roomIDChangedArgsForCall)
+}
+
+func (fake *FakeTelemetryService) RoomIDChangedCalls(stub func(context.Context, livekit.RoomID, *livekit.Room)) {
+	fake.roomIDChangedMutex.Lock()
+	defer fake.roomIDChangedMutex.Unlock()
+	fake.RoomIDChangedStub = stub
+}
+
+func (fake *FakeTelemetryService) RoomIDChangedArgsForCall(i int) (context.Context, livekit.RoomID, *livekit.Room) {
+	fake.roomIDChangedMutex.RLock()
+	defer fake.roomIDChangedMutex.RUnlock()
+	argsForCall := fake.roomIDChangedArgsForCall[i]
+	return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3
+}
+
 func (fake *FakeTelemetryService) RoomProjectReporter(arg1 context.Context) roomobs.ProjectReporter {
 	fake.roomProjectReporterMutex.Lock()
 	ret, specificReturn := fake.roomProjectReporterReturnsOnCall[len(fake.roomProjectReporterArgsForCall)]

pkg/telemetry/telemetryservice.go

@@ -19,6 +19,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/livekit/livekit-server/pkg/telemetry/prometheus"
 	"github.com/livekit/livekit-server/pkg/utils"
 	"github.com/livekit/protocol/codecs/mime"
 	"github.com/livekit/protocol/livekit"
@@ -45,6 +46,9 @@ type TelemetryService interface {
 	ParticipantResumed(ctx context.Context, room *livekit.Room, participant *livekit.ParticipantInfo, nodeID livekit.NodeID, reason livekit.ReconnectReason)
 	// ParticipantLeft - the participant leaves the room, only sent if ParticipantActive has been called before
 	ParticipantLeft(ctx context.Context, room *livekit.Room, participant *livekit.ParticipantInfo, shouldSendEvent bool, guard *ReferenceGuard)
+	// RoomIDChanged - the room kept its session, but got a different id (a provisional room id
+	// replaced by the resolved one), re-keys the stats workers of every participant in the room
+	RoomIDChanged(ctx context.Context, prevRoomID livekit.RoomID, room *livekit.Room)
 	// TrackPublishRequested - a publication attempt has been received
 	TrackPublishRequested(ctx context.Context, room *livekit.Room, participantID livekit.ParticipantID, identity livekit.ParticipantIdentity, track *livekit.TrackInfo, shouldSendEvent bool)
 	// TrackPublished - a publication attempt has been successful
@@ -115,6 +119,8 @@ func (n NullTelemetryService) ParticipantResumed(ctx context.Context, room *live
 }
 func (n NullTelemetryService) ParticipantLeft(ctx context.Context, room *livekit.Room, participant *livekit.ParticipantInfo, shouldSendEvent bool, guard *ReferenceGuard) {
 }
+func (n NullTelemetryService) RoomIDChanged(ctx context.Context, prevRoomID livekit.RoomID, room *livekit.Room) {
+}
 func (n NullTelemetryService) TrackPublishRequested(ctx context.Context, room *livekit.Room, participantID livekit.ParticipantID, identity livekit.ParticipantIdentity, track *livekit.TrackInfo, shouldSendEvent bool) {
 }
 func (n NullTelemetryService) TrackPublished(ctx context.Context, room *livekit.Room, participantID livekit.ParticipantID, identity livekit.ParticipantIdentity, track *livekit.TrackInfo, shouldSendEvent bool) {
@@ -166,11 +172,6 @@ const (
 	telemetryStatsUpdateInterval = time.Second * 30
 )
 
-type statsWorkerKey struct {
-	roomID        livekit.RoomID
-	participantID livekit.ParticipantID
-}
-
 type telemetryService struct {
 	AnalyticsService
 
@@ -178,7 +179,7 @@ type telemetryService struct {
 	jobsQueue *utils.OpsQueue
 
 	workersMu  sync.RWMutex
-	workers    map[statsWorkerKey]*StatsWorker
+	workers    map[livekit.RoomID]map[livekit.ParticipantID]*StatsWorker
 	workerList *StatsWorker
 
 	flushMu sync.Mutex
@@ -194,7 +195,7 @@ func NewTelemetryService(notifier webhook.QueuedNotifier, analytics AnalyticsSer
 			FlushOnStop: true,
 			Logger:      logger.GetLogger(),
 		}),
-		workers: make(map[statsWorkerKey]*StatsWorker),
+		workers: make(map[livekit.RoomID]map[livekit.ParticipantID]*StatsWorker),
 	}
 
 	t.jobsQueue.Start()
@@ -243,9 +244,12 @@ func (t *telemetryService) FlushStats() {
 	if reap != nil {
 		t.workersMu.Lock()
 		for reap != nil {
-			key := statsWorkerKey{reap.roomID, reap.participantID}
-			if reap == t.workers[key] {
-				delete(t.workers, key)
+			roomID := reap.RoomID()
+			if roomWorkers := t.workers[roomID]; reap == roomWorkers[reap.participantID] {
+				delete(roomWorkers, reap.participantID)
+				if len(roomWorkers) == 0 {
+					delete(t.workers, roomID)
+				}
 			}
 			reap = reap.next
 		}
@@ -267,7 +271,7 @@ func (t *telemetryService) getWorker(roomID livekit.RoomID, participantID liveki
 	t.workersMu.RLock()
 	defer t.workersMu.RUnlock()
 
-	worker, ok = t.workers[statsWorkerKey{roomID, participantID}]
+	worker, ok = t.workers[roomID][participantID]
 	return
 }
 
@@ -282,8 +286,8 @@ func (t *telemetryService) getOrCreateWorker(
 	t.workersMu.Lock()
 	defer t.workersMu.Unlock()
 
-	key := statsWorkerKey{roomID, participantID}
-	worker, ok := t.workers[key]
+	roomWorkers := t.workers[roomID]
+	worker, ok := roomWorkers[participantID]
 	if ok && !worker.Closed(guard) {
 		return worker, true
 	}
@@ -306,7 +310,11 @@ func (t *telemetryService) getOrCreateWorker(
 		worker.SetConnected()
 	}
 
-	t.workers[key] = worker
+	if roomWorkers == nil {
+		roomWorkers = make(map[livekit.ParticipantID]*StatsWorker)
+		t.workers[roomID] = roomWorkers
+	}
+	roomWorkers[participantID] = worker
 
 	worker.next = t.workerList
 	t.workerList = worker
@@ -314,6 +322,70 @@ func (t *telemetryService) getOrCreateWorker(
 	return worker, false
 }
 
+// reKeyRoom files every one of a room's stats workers under `roomID` instead of
+// `prevRoomID`.
+//
+// A room can be restarted while participants are connected and reporting stats, which
+// gives it a new id. As every worker of the room moves at once, the move is a single map
+// splice - the workers themselves are untouched and keep their place in the flush list.
+// Each worker then seals off what it collected under `prevRoomID` so those stats stay
+// attributed to the session that ended (see StatsWorker.SetRoom).
+func (t *telemetryService) reKeyRoom(prevRoomID livekit.RoomID, roomID livekit.RoomID, roomName livekit.RoomName) {
+	if prevRoomID == roomID {
+		return
+	}
+
+	t.workersMu.Lock()
+	defer t.workersMu.Unlock()
+
+	roomWorkers := t.workers[prevRoomID]
+	if len(roomWorkers) == 0 {
+		delete(t.workers, prevRoomID)
+		return
+	}
+	delete(t.workers, prevRoomID)
+
+	existing := t.workers[roomID]
+	if existing == nil {
+		t.workers[roomID] = roomWorkers
+	} else {
+		// should not happen as a room id is only ever replaced by a freshly minted one
+		logger.Warnw(
+			"telemetry re-keying room into an existing entry", nil,
+			"prevRoomID", prevRoomID,
+			"room", roomName,
+			"roomID", roomID,
+			"numWorkers", len(roomWorkers),
+			"numExistingWorkers", len(existing),
+		)
+	}
+
+	for participantID, worker := range roomWorkers {
+		if existing != nil {
+			if survivor, ok := existing[participantID]; ok {
+				// only one worker can be keyed at (room, participant) and the one already
+				// filed there wins, close the superseded one so that it drains and is
+				// reaped instead of lingering in the flush list unreachable
+				if worker.ForceClose(survivor) {
+					prometheus.SubParticipant()
+				}
+				continue
+			}
+			existing[participantID] = worker
+		}
+
+		worker.SetRoom(roomID, roomName)
+	}
+
+	logger.Infow(
+		"telemetry re-keyed room",
+		"prevRoomID", prevRoomID,
+		"room", roomName,
+		"roomID", roomID,
+		"numWorkers", len(roomWorkers),
+	)
+}
+
 func (t *telemetryService) LocalRoomState(ctx context.Context, info *livekit.AnalyticsNodeRooms) {
 	t.enqueue(func() {
 		t.SendNodeRoomStates(ctx, info)