From 6c3a5ae72a3db5607351e0bf17cc01bd2ae23741 Mon Sep 17 00:00:00 2001 From: Genseric Ghiro Date: Thu, 6 Aug 2026 17:58:14 -0400 Subject: [PATCH 1/5] Resubscribing to all remote tracks on rejoin --- pkg/sip/outbound_utilities_test.go | 8 +- pkg/sip/room.go | 288 ++++++++++++++++++++++------- 2 files changed, 230 insertions(+), 66 deletions(-) diff --git a/pkg/sip/outbound_utilities_test.go b/pkg/sip/outbound_utilities_test.go index e678167ae..73fa679a6 100644 --- a/pkg/sip/outbound_utilities_test.go +++ b/pkg/sip/outbound_utilities_test.go @@ -159,12 +159,12 @@ func newTestRoomWithConfig(log logger.Logger, st *RoomStats, cfg *testRoomConfig }) // Set up minimal participant info - room.p = ParticipantInfo{ + room.p.Store(&ParticipantInfo{ ID: "test-participant-id", RoomName: "test-room", Identity: "test-participant", Name: "Test Participant", - } + }) return &testRoom{room: room} } @@ -173,11 +173,11 @@ func newTestRoomWithConfig(log logger.Logger, st *RoomStats, cfg *testRoomConfig func (r *testRoom) Connect(_ context.Context, conf *config.Config, rconf RoomConfig) error { // Update participant info from config partConf := rconf.Participant - r.room.p = ParticipantInfo{ + r.room.p.Store(&ParticipantInfo{ RoomName: rconf.RoomName, Identity: partConf.Identity, Name: partConf.Name, - } + }) // Skip actual connection - room is already set up return nil } diff --git a/pkg/sip/room.go b/pkg/sip/room.go index 4c0024fd0..116306dcd 100644 --- a/pkg/sip/room.go +++ b/pkg/sip/room.go @@ -31,14 +31,12 @@ import ( "github.com/livekit/media-sdk/dtmf" "github.com/livekit/media-sdk/g711" "github.com/livekit/media-sdk/jitter" + "github.com/livekit/media-sdk/mixer" "github.com/livekit/media-sdk/rtp" "github.com/livekit/protocol/livekit" "github.com/livekit/protocol/logger" "github.com/livekit/protocol/sip" lksdk "github.com/livekit/server-sdk-go/v2" - - "github.com/livekit/media-sdk/mixer" - "github.com/livekit/sip/pkg/config" "github.com/livekit/sip/pkg/media/opus" ) @@ -65,6 +63,13 @@ type RoomStatsSnapshot struct { JitterBufferPacketsLost uint64 `json:"jitter_buffer_packets_lost"` JitterBufferPacketsDropped uint64 `json:"jitter_buffer_packets_dropped"` + TrackSubscribes uint64 `json:"track_subscribes"` + Reconnects uint64 `json:"reconnects"` + Rejoins uint64 `json:"rejoins"` + // Reconnecting reports whether the signal connection was down when the + // snapshot was taken. PublishedFrames and PublishTX are unreliable while set. + Reconnecting bool `json:"reconnecting"` + LatencyOutRecv LatencyStatsSnapshot `json:"latency_out_recv"` Closed bool `json:"closed"` @@ -78,6 +83,21 @@ type RoomStats struct { rtpStats rtpCountingStats dataPackets atomic.Uint64 + // TrackSubscribes counts subscribe requests issued for remote tracks. + // Attempts, not confirmations. + TrackSubscribes atomic.Uint64 + + // Reconnects counts signal connections lost and recovered during this call. + // Rejoins is the subset that rebuilt the peer connections. + Reconnects atomic.Uint64 + Rejoins atomic.Uint64 + + // Reconnecting is set while the signal connection is down. PublishedFrames + // and PublishTX are counted before the track write, so they keep reporting a + // healthy rate even though the audio is being dropped. Read them only when + // this is false. + Reconnecting atomic.Bool + JitterBufferPacketsLost atomic.Uint64 JitterBufferPacketsDropped atomic.Uint64 @@ -110,6 +130,11 @@ func (s *RoomStats) Load() RoomStatsSnapshot { JitterBufferPacketsLost: s.JitterBufferPacketsLost.Load(), JitterBufferPacketsDropped: s.JitterBufferPacketsDropped.Load(), + TrackSubscribes: s.TrackSubscribes.Load(), + Reconnects: s.Reconnects.Load(), + Rejoins: s.Rejoins.Load(), + Reconnecting: s.Reconnecting.Load(), + PublishedFrames: s.PublishedFrames.Load(), PublishedSamples: s.PublishedSamples.Load(), PublishTX: math.Float64frombits(s.PublishTX.Load()), @@ -173,13 +198,16 @@ func DefaultGetRoomFunc(log logger.Logger, st *RoomStats) RoomInterface { } type Room struct { - log logger.Logger - roomLog logger.Logger // deferred logger - room *lksdk.Room - mix *mixer.Mixer - out *msdk.SwitchWriter - outDtmf atomic.Pointer[dtmf.Writer] - p ParticipantInfo + log logger.Logger + roomLog logger.Logger // deferred logger + room *lksdk.Room + mix *mixer.Mixer + out *msdk.SwitchWriter + outDtmf atomic.Pointer[dtmf.Writer] + // p is replaced on every full rejoin, since the server issues a new + // participant SID, and read concurrently by Participant(). + p atomic.Pointer[ParticipantInfo] + reconnect atomic.Pointer[reconnectState] ready core.Fuse subscribe atomic.Bool subscribed core.Fuse @@ -295,6 +323,7 @@ func (r *Room) subscribeTo(pub *lksdk.RemoteTrackPublication, rp *lksdk.RemotePa return } log.Debugw("subscribing to a track") + r.stats.TrackSubscribes.Add(1) if err := pub.SetSubscribed(true); err != nil { log.Errorw("cannot subscribe to the track", err) return @@ -307,12 +336,89 @@ func (r *Room) Connect(ctx context.Context, conf *config.Config, rconf RoomConfi rconf.WsUrl = conf.WsUrl } partConf := rconf.Participant - r.p = ParticipantInfo{ + r.p.Store(&ParticipantInfo{ RoomName: rconf.RoomName, Identity: partConf.Identity, Name: partConf.Name, + }) + roomCallback := r.newRoomCallback(conf, rconf) + + if rconf.Token == "" { + // TODO: Remove this code path, always sign tokens on LiveKit server. + // For now, match Cloud behavior and do not send extra attrs in the token. + tokenAttrs := make(map[string]string, len(partConf.Attributes)) + for _, k := range []string{ + livekit.AttrSIPCallID, + livekit.AttrSIPTrunkID, + livekit.AttrSIPDispatchRuleID, + livekit.AttrSIPTrunkNumber, + livekit.AttrSIPPhoneNumber, + } { + if v, ok := partConf.Attributes[k]; ok { + tokenAttrs[k] = v + } + } + var err error + rconf.Token, err = sip.BuildSIPToken(sip.SIPTokenParams{ + APIKey: conf.ApiKey, + APISecret: conf.ApiSecret, + RoomName: rconf.RoomName, + ParticipantIdentity: partConf.Identity, + ParticipantName: partConf.Name, + ParticipantMetadata: partConf.Metadata, + ParticipantAttributes: tokenAttrs, + RoomPreset: rconf.RoomPreset, + RoomConfig: rconf.RoomConfig, + }) + if err != nil { + return err + } + } + room := lksdk.NewRoom(roomCallback) + room.SetLogger(newRoomOverrideLogger(r.log)) + err := room.JoinWithContextAndToken(ctx, rconf.WsUrl, rconf.Token, + lksdk.WithAutoSubscribe(false), + lksdk.WithExtraAttributes(partConf.Attributes), + ) + if err != nil { + return err + } + r.room = room + r.setParticipantFromRoom() + p := r.Participant() + r.log = r.log.WithValues("room", room.Name(), "roomID", room.SID(), "participant", p.Identity, "participantID", p.ID) + r.log.Infow("SIP participant joined room") + room.LocalParticipant.SetAttributes(partConf.Attributes) + r.ready.Break() + r.subscribe.Store(false) // already false, but keep for visibility + + // Not subscribing to any tracks just yet! + return nil +} + +// setParticipantFromRoom refreshes the cached participant identifiers from the +// SDK room. Runs on reconnect too, since a full rejoin gets a new SID. +// +// Does not rebuild r.log, which is read without synchronisation elsewhere in +// this file. The reconnect handler logs the SID change instead. +func (r *Room) setParticipantFromRoom() { + room := r.room + if room == nil { + return } - roomCallback := &lksdk.RoomCallback{ + p := ParticipantInfo{} + if cur := r.p.Load(); cur != nil { + p = *cur + } + p.ID = room.LocalParticipant.SID() + p.Identity = room.LocalParticipant.Identity() + r.p.Store(&p) +} + +// newRoomCallback builds the LiveKit room callback for this SIP participant. +// Separate from Connect so tests can build it without joining a room. +func (r *Room) newRoomCallback(conf *config.Config, rconf RoomConfig) *lksdk.RoomCallback { + return &lksdk.RoomCallback{ OnParticipantConnected: func(rp *lksdk.RemoteParticipant) { log := r.roomLog.WithValues("participant", rp.Identity(), "participantID", rp.SID()) if !r.subscribe.Load() { @@ -414,62 +520,108 @@ func (r *Room) Connect(ctx context.Context, conf *config.Config, rconf RoomConfi r.roomLog.Infow("track unsubscribed", "participant", rp.Identity(), "participantID", rp.SID(), "trackID", track.ID(), "trackName", pub.Name()) }, }, + OnReconnecting: func() { + r.onReconnecting() + }, + OnReconnected: func() { + r.onReconnected() + }, OnDisconnected: func() { r.stopped.Break() }, + OnDisconnectedWithReason: func(reason lksdk.DisconnectionReason) { + // OnDisconnected fires first and owns the teardown. This only + // records the reason, which CloseWithReason may clear later. + r.roomLog.Infow("disconnected from room", "reason", reason) + }, } +} - if rconf.Token == "" { - // TODO: Remove this code path, always sign tokens on LiveKit server. - // For now, match Cloud behavior and do not send extra attrs in the token. - tokenAttrs := make(map[string]string, len(partConf.Attributes)) - for _, k := range []string{ - livekit.AttrSIPCallID, - livekit.AttrSIPTrunkID, - livekit.AttrSIPDispatchRuleID, - livekit.AttrSIPTrunkNumber, - livekit.AttrSIPPhoneNumber, - } { - if v, ok := partConf.Attributes[k]; ok { - tokenAttrs[k] = v - } - } - var err error - rconf.Token, err = sip.BuildSIPToken(sip.SIPTokenParams{ - APIKey: conf.ApiKey, - APISecret: conf.ApiSecret, - RoomName: rconf.RoomName, - ParticipantIdentity: partConf.Identity, - ParticipantName: partConf.Name, - ParticipantMetadata: partConf.Metadata, - ParticipantAttributes: tokenAttrs, - RoomPreset: rconf.RoomPreset, - RoomConfig: rconf.RoomConfig, - }) - if err != nil { - return err - } +// reconnectState is captured when the signal connection drops so onReconnected +// can tell a resume from a full rejoin and report the gap. +type reconnectState struct { + startedAt time.Time + sid string +} + +// onReconnecting runs when the SDK loses the signal connection and starts +// recovering. Audio published until onReconnected may be dropped: a full rejoin +// detaches our track from its peer connection while the connection is rebuilt, +// and writes to a detached track are discarded without an error. +func (r *Room) onReconnecting() { + var sid string + if room := r.room; room != nil { + sid = room.LocalParticipant.SID() } - room := lksdk.NewRoom(roomCallback) - room.SetLogger(newRoomOverrideLogger(r.log)) - err := room.JoinWithContextAndToken(ctx, rconf.WsUrl, rconf.Token, - lksdk.WithAutoSubscribe(false), - lksdk.WithExtraAttributes(partConf.Attributes), + r.reconnect.Store(&reconnectState{startedAt: time.Now(), sid: sid}) + r.stats.Reconnecting.Store(true) + r.stats.Reconnects.Add(1) + r.roomLog.Infow("lost connection to room, reconnecting", "participantID", sid) +} + +// onReconnected runs when the SDK recovers the signal connection, either by +// resuming the old session or rejoining from scratch. +// +// A resume keeps the peer connections, and the SDK replays subscription state +// itself, so leave it alone. A full rejoin builds a new subscriber peer +// connection with no subscriptions and the SDK restores only what we publish, +// so re-issue the subscriptions here. +func (r *Room) onReconnected() { + prev := r.reconnect.Swap(nil) + r.stats.Reconnecting.Store(false) + + room := r.room + if room == nil { + return + } + + // The SID is stable across a resume and changes on a full rejoin. Treat an + // unknown previous SID as a rejoin: re-subscribing is idempotent, while + // missing one leaves the call with no inbound room audio. + sid := room.LocalParticipant.SID() + rejoined := prev == nil || prev.sid != sid + + var gap time.Duration + if prev != nil { + gap = time.Since(prev.startedAt) + } + if rejoined { + r.stats.Rejoins.Add(1) + } + + r.setParticipantFromRoom() + r.roomLog.Infow("reconnected to room", + "rejoined", rejoined, + "gap", gap, + "participantID", sid, + "previousParticipantID", func() string { + if prev == nil { + return "" + } + return prev.sid + }(), ) - if err != nil { - return err + + if !rejoined { + return } - r.room = room - r.p.ID = r.room.LocalParticipant.SID() - r.p.Identity = r.room.LocalParticipant.Identity() - r.log = r.log.WithValues("room", r.room.Name(), "roomID", r.room.SID(), "participant", r.p.Identity, "participantID", r.p.ID) - r.log.Infow("SIP participant joined room") - room.LocalParticipant.SetAttributes(partConf.Attributes) - r.ready.Break() - r.subscribe.Store(false) // already false, but keep for visibility + if !r.subscribe.Load() { + // Call is not answered yet, so subscribing here would pull room audio + // into a leg that has not been accepted. + return + } + // The SDK calls this from the rejoin itself, inside the join's timeout, and + // subscribing does a blocking websocket write per track. Pass the room we + // already read so a concurrent close cannot make this a nil dereference. + go r.resubscribeAfterRejoin(room) +} - // Not subscribing to any tracks just yet! - return nil +func (r *Room) resubscribeAfterRejoin(room *lksdk.Room) { + if r.closed.IsBroken() || r.stopped.IsBroken() { + return + } + r.roomLog.Infow("re-subscribing to remote tracks after rejoin") + r.subscribeAll(room) } func (r *Room) RegisterRpcCtxMethod(method string, handler lksdk.RpcHandlerCtxFunc) error { @@ -477,11 +629,20 @@ func (r *Room) RegisterRpcCtxMethod(method string, handler lksdk.RpcHandlerCtxFu } func (r *Room) Subscribe() { - if r.room == nil { + // CloseWithReason clears r.room from another goroutine. Copy it first so the + // nil check and the use below see the same value. + room := r.room + if room == nil { return } r.subscribe.Store(true) - list := r.room.GetRemoteParticipants() + r.subscribeAll(room) +} + +// subscribeAll subscribes to every remote audio track in the room. Safe to +// repeat, since a duplicate subscribe is a no-op server side. +func (r *Room) subscribeAll(room *lksdk.Room) { + list := room.GetRemoteParticipants() r.log.Debugw("subscribing to existing room participants", "participants", len(list)) for _, rp := range list { r.participantJoin(rp) @@ -571,7 +732,10 @@ func (r *Room) Participant() ParticipantInfo { if r == nil { return ParticipantInfo{} } - return r.p + if p := r.p.Load(); p != nil { + return *p + } + return ParticipantInfo{} } func (r *Room) NewParticipantTrack(sampleRate int) (msdk.WriteCloser[msdk.PCM16Sample], error) { From b9737a64daa6aafb10cdb520f1d157d0a84b8fa7 Mon Sep 17 00:00:00 2001 From: Genseric Ghiro Date: Fri, 7 Aug 2026 14:46:00 -0400 Subject: [PATCH 2/5] Making room an atomic field --- pkg/sip/outbound_utilities_test.go | 5 +-- pkg/sip/room.go | 56 ++++++++++++++++++++---------- 2 files changed, 40 insertions(+), 21 deletions(-) diff --git a/pkg/sip/outbound_utilities_test.go b/pkg/sip/outbound_utilities_test.go index 73fa679a6..785508075 100644 --- a/pkg/sip/outbound_utilities_test.go +++ b/pkg/sip/outbound_utilities_test.go @@ -141,7 +141,8 @@ func newTestRoomWithConfig(log logger.Logger, st *RoomStats, cfg *testRoomConfig room.roomLog = roomLog // Create a minimal lksdk.Room without connecting - room.room = lksdk.NewRoom(nil) + sdkRoom := lksdk.NewRoom(nil) + room.room.Store(sdkRoom) // Set ready immediately (skip connection) room.ready.Break() @@ -150,7 +151,7 @@ func newTestRoomWithConfig(log logger.Logger, st *RoomStats, cfg *testRoomConfig } resolve.Resolve() - room.room.OnRoomUpdate(&livekit.Room{ // Set metadata, and specifically Sid + sdkRoom.OnRoomUpdate(&livekit.Room{ // Set metadata, and specifically Sid Name: "test-room", Metadata: "test-metadata", Sid: "test-room-sid", diff --git a/pkg/sip/room.go b/pkg/sip/room.go index 116306dcd..99cdf615e 100644 --- a/pkg/sip/room.go +++ b/pkg/sip/room.go @@ -41,6 +41,10 @@ import ( "github.com/livekit/sip/pkg/media/opus" ) +// errRoomClosed is returned when the room handle is gone, which happens once the +// call has been torn down. +var errRoomClosed = errors.New("room is closed") + type RoomStatsSnapshot struct { // Stats quantifying total incoming traffic from all tracks InputPackets uint64 `json:"input_packets"` @@ -200,7 +204,8 @@ func DefaultGetRoomFunc(log logger.Logger, st *RoomStats) RoomInterface { type Room struct { log logger.Logger roomLog logger.Logger // deferred logger - room *lksdk.Room + // room is cleared on close while SDK callback goroutines still read it. + room atomic.Pointer[lksdk.Room] mix *mixer.Mixer out *msdk.SwitchWriter outDtmf atomic.Pointer[dtmf.Writer] @@ -252,8 +257,8 @@ func NewRoom(log logger.Logger, st *RoomStats) *Room { go func() { select { case <-r.ready.Watch(): - if r.room != nil { - resolve.Resolve("room", r.room.Name(), "roomID", r.room.SID()) + if room := r.room.Load(); room != nil { + resolve.Resolve("room", room.Name(), "roomID", room.SID()) } else { resolve.Resolve() } @@ -278,10 +283,14 @@ func (r *Room) Closed() <-chan struct{} { // fired. Returns livekit.DisconnectReason_UNKNOWN_REASON if the room hasn't // disconnected or no reason was reported. func (r *Room) ClosedReason() livekit.DisconnectReason { - if r == nil || r.room == nil { + if r == nil { + return livekit.DisconnectReason_UNKNOWN_REASON + } + room := r.room.Load() + if room == nil { return livekit.DisconnectReason_UNKNOWN_REASON } - return r.room.DisconnectReason() + return room.DisconnectReason() } func (r *Room) Subscribed() <-chan struct{} { @@ -295,7 +304,7 @@ func (r *Room) Room() *lksdk.Room { if r == nil { return nil } - return r.room + return r.room.Load() } func (r *Room) participantJoin(rp *lksdk.RemoteParticipant) { @@ -383,7 +392,7 @@ func (r *Room) Connect(ctx context.Context, conf *config.Config, rconf RoomConfi if err != nil { return err } - r.room = room + r.room.Store(room) r.setParticipantFromRoom() p := r.Participant() r.log = r.log.WithValues("room", room.Name(), "roomID", room.SID(), "participant", p.Identity, "participantID", p.ID) @@ -402,7 +411,7 @@ func (r *Room) Connect(ctx context.Context, conf *config.Config, rconf RoomConfi // Does not rebuild r.log, which is read without synchronisation elsewhere in // this file. The reconnect handler logs the SID change instead. func (r *Room) setParticipantFromRoom() { - room := r.room + room := r.room.Load() if room == nil { return } @@ -550,7 +559,7 @@ type reconnectState struct { // and writes to a detached track are discarded without an error. func (r *Room) onReconnecting() { var sid string - if room := r.room; room != nil { + if room := r.room.Load(); room != nil { sid = room.LocalParticipant.SID() } r.reconnect.Store(&reconnectState{startedAt: time.Now(), sid: sid}) @@ -570,7 +579,7 @@ func (r *Room) onReconnected() { prev := r.reconnect.Swap(nil) r.stats.Reconnecting.Store(false) - room := r.room + room := r.room.Load() if room == nil { return } @@ -625,13 +634,15 @@ func (r *Room) resubscribeAfterRejoin(room *lksdk.Room) { } func (r *Room) RegisterRpcCtxMethod(method string, handler lksdk.RpcHandlerCtxFunc) error { - return r.room.RegisterRpcCtxMethod(method, handler) + room := r.room.Load() + if room == nil { + return errRoomClosed + } + return room.RegisterRpcCtxMethod(method, handler) } func (r *Room) Subscribe() { - // CloseWithReason clears r.room from another goroutine. Copy it first so the - // nil check and the use below see the same value. - room := r.room + room := r.room.Load() if room == nil { return } @@ -717,9 +728,8 @@ func (r *Room) CloseWithReason(reason livekit.DisconnectReason) error { r.subscribe.Store(false) err = r.CloseOutput() r.SetDTMFOutput(nil) - if r.room != nil { - r.room.DisconnectWithReason(reason) - r.room = nil + if room := r.room.Swap(nil); room != nil { + room.DisconnectWithReason(reason) } if r.mix != nil { r.mix.Stop() @@ -743,7 +753,11 @@ func (r *Room) NewParticipantTrack(sampleRate int) (msdk.WriteCloser[msdk.PCM16S if err != nil { return nil, err } - p := r.room.LocalParticipant + room := r.room.Load() + if room == nil { + return nil, errRoomClosed + } + p := room.LocalParticipant if _, err = p.PublishTrack(track, &lksdk.TrackPublicationOptions{ Name: p.Identity(), }); err != nil { @@ -761,7 +775,11 @@ func (r *Room) SendData(data lksdk.DataPacket, opts ...lksdk.DataPublishOption) if r == nil || !r.ready.IsBroken() || r.closed.IsBroken() { return nil } - return r.room.LocalParticipant.PublishDataPacket(data, opts...) + room := r.room.Load() + if room == nil { + return nil + } + return room.LocalParticipant.PublishDataPacket(data, opts...) } func (r *Room) NewTrack() *mixer.Input { From 1a3e33185f6c390d569aa20ad750906a243d1dac Mon Sep 17 00:00:00 2001 From: Genseric Ghiro Date: Fri, 7 Aug 2026 14:46:10 -0400 Subject: [PATCH 3/5] Tests --- pkg/sip/room_test.go | 282 +++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 282 insertions(+) create mode 100644 pkg/sip/room_test.go diff --git a/pkg/sip/room_test.go b/pkg/sip/room_test.go new file mode 100644 index 000000000..2ed6b1dc4 --- /dev/null +++ b/pkg/sip/room_test.go @@ -0,0 +1,282 @@ +// Copyright 2026 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 sip + +import ( + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "github.com/livekit/protocol/livekit" + "github.com/livekit/protocol/logger" + lksdk "github.com/livekit/server-sdk-go/v2" + + "github.com/livekit/sip/pkg/config" +) + +const ( + testRemoteIdentity = "agent" + testRemoteSID = "PA_remote" + testRemoteTrackSID = "TR_remote_audio" +) + +func testRoomInfo() *livekit.Room { + return &livekit.Room{Sid: "RM_test", Name: "test-room"} +} + +func testLocalInfo(sid string) *livekit.ParticipantInfo { + return &livekit.ParticipantInfo{ + Sid: sid, + Identity: "sip-participant", + Kind: livekit.ParticipantInfo_SIP, + } +} + +// testRemoteInfo is the other party in the room, holding one audio track. It +// never reconnects, so its SIDs stay the same across our rejoin, which is what +// makes the SDK treat its track as already known. +func testRemoteInfo() []*livekit.ParticipantInfo { + return []*livekit.ParticipantInfo{{ + Sid: testRemoteSID, + Identity: testRemoteIdentity, + State: livekit.ParticipantInfo_ACTIVE, + Tracks: []*livekit.TrackInfo{{ + Sid: testRemoteTrackSID, + Type: livekit.TrackType_AUDIO, + Name: "microphone", + }}, + }} +} + +// --- reconnect --------------------------------------------------------------- + +type reconnectFixture struct { + room *Room + sdk *lksdk.Room + published *atomic.Int32 // times SIP's OnTrackPublished handler ran +} + +func newReconnectFixture(t *testing.T) *reconnectFixture { + t.Helper() + + r := NewRoom(logger.GetLogger(), &RoomStats{}) + t.Cleanup(func() { _ = r.Close() }) + + cb := r.newRoomCallback(&config.Config{}, RoomConfig{}) + + // Wrap the callback before handing it to the SDK: NewRoom copies the fields + // via Merge, so wrapping afterwards would not be observed. + var published atomic.Int32 + inner := cb.ParticipantCallback.OnTrackPublished + cb.ParticipantCallback.OnTrackPublished = func(pub *lksdk.RemoteTrackPublication, rp *lksdk.RemoteParticipant) { + published.Add(1) + inner(pub, rp) + } + + sdk := lksdk.NewRoom(cb) + r.room.Store(sdk) + r.ready.Break() + + return &reconnectFixture{room: r, sdk: sdk, published: &published} +} + +// join brings the fixture to the state of an answered call: joined, remote +// participant present, and subscribing enabled. +func (f *reconnectFixture) join(t *testing.T) { + t.Helper() + + f.sdk.OnRoomJoined(testRoomInfo(), testLocalInfo("PA_sip_1"), testRemoteInfo(), &livekit.ServerInfo{}, nil) + require.EqualValues(t, 1, f.published.Load(), "expected the initial publication to be announced") + + f.room.Subscribe() + require.True(t, f.room.subscribe.Load()) + require.Len(t, f.sdk.GetRemoteParticipants(), 1) + + f.published.Store(0) // only count what happens from here on +} + +// rejoinEscalated simulates a failed resume escalating to a full rejoin, where +// the SDK skips OnRestarting. +func (f *reconnectFixture) rejoinEscalated() { + f.sdk.OnResuming() + f.sdk.OnRoomJoined(testRoomInfo(), testLocalInfo("PA_sip_2"), testRemoteInfo(), &livekit.ServerInfo{}, nil) + f.sdk.OnRestarted(testRoomInfo(), testLocalInfo("PA_sip_2"), testRemoteInfo()) +} + +// rejoinServerInitiated simulates the server asking for a full reconnect, where +// OnRestarting does run. +func (f *reconnectFixture) rejoinServerInitiated() { + f.sdk.OnRestarting() + f.sdk.OnRoomJoined(testRoomInfo(), testLocalInfo("PA_sip_2"), testRemoteInfo(), &livekit.ServerInfo{}, nil) + f.sdk.OnRestarted(testRoomInfo(), testLocalInfo("PA_sip_2"), testRemoteInfo()) +} + +// TestRoomReconnect covers what happens to a call when the SIP pod loses +// its signal connection to the server and recovers it. +// +// These subtests call the SDK's exported reconnect handlers directly. A real +// reconnect needs a live peer connection to succeed, so a fake signal server +// would have to complete ICE/DTLS to reach the same states. +// +// Only participants connected to the affected pod reconnect. Everyone other +// participant in that room is unaffected and keep their session and their SIDs. +// So "clearing participants" below means dropping our own view of them, not +// removing anyone from the room. There are two ways to reach a full rejoin: +// +// 1- server-initiated: rejoin is attempted right away, participants affected +// drop their participant map, and then rebuild it from the OnRoomJoined +// snapshot and re-announce their tracks. +// 2- resume-then-escalate: resume is first attempted. If it fails, then switch +// to a rejoin, and the stale participant map survives and no tracks are announced. +func TestRoomReconnect(t *testing.T) { + t.Run("escalated rejoin does not re-announce tracks from remote participants", func(t *testing.T) { + f := newReconnectFixture(t) + f.join(t) + + f.rejoinEscalated() + + require.EqualValues(t, 0, f.published.Load(), + "OnTrackPublished must not re-fire on the escalated path") + }) + + // The other path does re-announce, which is why lost audio is intermittent. + t.Run("server initiated rejoin re-announces tracks from remote participants", func(t *testing.T) { + f := newReconnectFixture(t) + f.join(t) + + f.rejoinServerInitiated() + + require.EqualValues(t, 1, f.published.Load(), + "OnTrackPublished is expected to re-fire when OnRestarting cleared the participants") + }) + + // However the rejoin was reached, SIP must re-issue its subscriptions. + t.Run("resubscribes after escalated rejoin", func(t *testing.T) { + f := newReconnectFixture(t) + f.join(t) + + before := f.room.stats.TrackSubscribes.Load() + f.rejoinEscalated() + + require.Eventually(t, func() bool { + return f.room.stats.TrackSubscribes.Load() > before + }, time.Second, 10*time.Millisecond, + "SIP must re-subscribe to remote tracks after a full rejoin") + }) + + // An outbound call joins the room and publishes before it starts dialing, but + // defers Subscribe() until the callee answers. A reconnect in that window + // must not start pulling room audio toward a leg nobody has picked up. + t.Run("does not subscribe before the call is answered", func(t *testing.T) { + f := newReconnectFixture(t) + + // Joined, but Subscribe() has not been called yet. + f.sdk.OnRoomJoined(testRoomInfo(), testLocalInfo("PA_sip_1"), testRemoteInfo(), &livekit.ServerInfo{}, nil) + require.False(t, f.room.subscribe.Load()) + + before := f.room.stats.TrackSubscribes.Load() + f.rejoinEscalated() + + require.Never(t, func() bool { + return f.room.stats.TrackSubscribes.Load() > before + }, 200*time.Millisecond, 20*time.Millisecond, + "reconnect before answer must not subscribe") + require.False(t, f.room.subscribe.Load(), "reconnect must not flip the subscribe flag") + }) + + // A resume keeps its subscriptions and the SDK replays them itself. + t.Run("resume does not resubscribe", func(t *testing.T) { + f := newReconnectFixture(t) + f.join(t) + + before := f.room.stats.TrackSubscribes.Load() + f.sdk.OnResuming() + f.sdk.OnResumed() + + require.Never(t, func() bool { + return f.room.stats.TrackSubscribes.Load() > before + }, 200*time.Millisecond, 20*time.Millisecond, + "re-subscribing on resume would race the SDK's own sendSyncState") + }) + + t.Run("counts rejoins separately from resumes", func(t *testing.T) { + f := newReconnectFixture(t) + f.join(t) + + f.rejoinEscalated() + require.EqualValues(t, 1, f.room.stats.Rejoins.Load()) + + f.sdk.OnResuming() + f.sdk.OnResumed() + require.EqualValues(t, 2, f.room.stats.Reconnects.Load()) + require.EqualValues(t, 1, f.room.stats.Rejoins.Load(), "a resume must not count as a rejoin") + require.False(t, f.room.stats.Reconnecting.Load()) + }) + + // PublishedFrames and PublishTX keep climbing during a gap whether or not + // audio reaches the room, so the gap itself has to be visible. + t.Run("snapshot exposes the gap", func(t *testing.T) { + f := newReconnectFixture(t) + f.join(t) + + f.sdk.OnResuming() + require.True(t, f.room.stats.Load().Reconnecting, "gap must be visible while it is happening") + + f.sdk.OnResumed() + require.False(t, f.room.stats.Load().Reconnecting) + require.EqualValues(t, 1, f.room.stats.Load().Reconnects) + }) + + // A SIP leg can hang up at any point, including mid-recovery, so teardown + // runs concurrently with the reconnect handlers. This is a smoke test for + // that overlap, not a race guard: the SDK locks the two goroutines take + // incidentally order them, so -race does not reliably see the field access. + t.Run("survives teardown during recovery", func(t *testing.T) { + f := newReconnectFixture(t) + f.join(t) + + var wg sync.WaitGroup + wg.Add(2) + go func() { + defer wg.Done() + f.rejoinEscalated() + }() + go func() { + defer wg.Done() + _ = f.room.CloseWithReason(livekit.DisconnectReason_CLIENT_INITIATED) + }() + wg.Wait() + + require.Nil(t, f.room.Room(), "close must clear the room handle") + }) + + // The SID changes on every full rejoin and feeds call state. + t.Run("refreshes participant SID after rejoin", func(t *testing.T) { + f := newReconnectFixture(t) + f.join(t) + f.room.setParticipantFromRoom() + require.Equal(t, "PA_sip_1", f.room.Participant().ID) + + f.rejoinEscalated() + + require.Eventually(t, func() bool { + return f.room.Participant().ID == "PA_sip_2" + }, time.Second, 10*time.Millisecond, + "cached participant SID must be refreshed after a rejoin") + }) +} From db01d2c95c73a0615a04ce8e4fd53ed95617dcad Mon Sep 17 00:00:00 2001 From: Genseric Ghiro Date: Fri, 7 Aug 2026 15:04:28 -0400 Subject: [PATCH 4/5] nit: making the distinction between reconnections and rejoins clearer --- pkg/sip/room.go | 76 +++++++++++++++++++++++++------------------- pkg/sip/room_test.go | 69 +++++++++++++++++++++------------------- 2 files changed, 79 insertions(+), 66 deletions(-) diff --git a/pkg/sip/room.go b/pkg/sip/room.go index 99cdf615e..4b432967a 100644 --- a/pkg/sip/room.go +++ b/pkg/sip/room.go @@ -68,11 +68,11 @@ type RoomStatsSnapshot struct { JitterBufferPacketsDropped uint64 `json:"jitter_buffer_packets_dropped"` TrackSubscribes uint64 `json:"track_subscribes"` + Resumes uint64 `json:"resumes"` Reconnects uint64 `json:"reconnects"` - Rejoins uint64 `json:"rejoins"` - // Reconnecting reports whether the signal connection was down when the + // Recovering reports whether the signal connection was down when the // snapshot was taken. PublishedFrames and PublishTX are unreliable while set. - Reconnecting bool `json:"reconnecting"` + Recovering bool `json:"recovering"` LatencyOutRecv LatencyStatsSnapshot `json:"latency_out_recv"` @@ -91,16 +91,18 @@ type RoomStats struct { // Attempts, not confirmations. TrackSubscribes atomic.Uint64 - // Reconnects counts signal connections lost and recovered during this call. - // Rejoins is the subset that rebuilt the peer connections. + // Resumes and Reconnects count the two ways the signal connection recovers + // during a call, and are mutually exclusive. A resume keeps the peer + // connections and subscriptions; a reconnect rebuilds them. Neither is + // counted until the recovery succeeds. + Resumes atomic.Uint64 Reconnects atomic.Uint64 - Rejoins atomic.Uint64 - // Reconnecting is set while the signal connection is down. PublishedFrames + // Recovering is set while the signal connection is down. PublishedFrames // and PublishTX are counted before the track write, so they keep reporting a // healthy rate even though the audio is being dropped. Read them only when // this is false. - Reconnecting atomic.Bool + Recovering atomic.Bool JitterBufferPacketsLost atomic.Uint64 JitterBufferPacketsDropped atomic.Uint64 @@ -135,9 +137,9 @@ func (s *RoomStats) Load() RoomStatsSnapshot { JitterBufferPacketsDropped: s.JitterBufferPacketsDropped.Load(), TrackSubscribes: s.TrackSubscribes.Load(), + Resumes: s.Resumes.Load(), Reconnects: s.Reconnects.Load(), - Rejoins: s.Rejoins.Load(), - Reconnecting: s.Reconnecting.Load(), + Recovering: s.Recovering.Load(), PublishedFrames: s.PublishedFrames.Load(), PublishedSamples: s.PublishedSamples.Load(), @@ -209,7 +211,7 @@ type Room struct { mix *mixer.Mixer out *msdk.SwitchWriter outDtmf atomic.Pointer[dtmf.Writer] - // p is replaced on every full rejoin, since the server issues a new + // p is replaced on every reconnect, since the server issues a new // participant SID, and read concurrently by Participant(). p atomic.Pointer[ParticipantInfo] reconnect atomic.Pointer[reconnectState] @@ -406,7 +408,7 @@ func (r *Room) Connect(ctx context.Context, conf *config.Config, rconf RoomConfi } // setParticipantFromRoom refreshes the cached participant identifiers from the -// SDK room. Runs on reconnect too, since a full rejoin gets a new SID. +// SDK room. Runs on recovery too, since a reconnect gets a new SID. // // Does not rebuild r.log, which is read without synchronisation elsewhere in // this file. The reconnect handler logs the SID change instead. @@ -547,14 +549,21 @@ func (r *Room) newRoomCallback(conf *config.Config, rconf RoomConfig) *lksdk.Roo } // reconnectState is captured when the signal connection drops so onReconnected -// can tell a resume from a full rejoin and report the gap. +// can tell a resume from a reconnect and report the gap. type reconnectState struct { startedAt time.Time sid string } +func recoveryKind(resumed bool) string { + if resumed { + return "resume" + } + return "reconnect" +} + // onReconnecting runs when the SDK loses the signal connection and starts -// recovering. Audio published until onReconnected may be dropped: a full rejoin +// recovering. Audio published until onReconnected may be dropped: a reconnect // detaches our track from its peer connection while the connection is rebuilt, // and writes to a detached track are discarded without an error. func (r *Room) onReconnecting() { @@ -563,44 +572,45 @@ func (r *Room) onReconnecting() { sid = room.LocalParticipant.SID() } r.reconnect.Store(&reconnectState{startedAt: time.Now(), sid: sid}) - r.stats.Reconnecting.Store(true) - r.stats.Reconnects.Add(1) - r.roomLog.Infow("lost connection to room, reconnecting", "participantID", sid) + r.stats.Recovering.Store(true) + r.roomLog.Infow("lost connection to room, recovering", "participantID", sid) } // onReconnected runs when the SDK recovers the signal connection, either by -// resuming the old session or rejoining from scratch. +// resuming the old session or reconnecting from scratch. // // A resume keeps the peer connections, and the SDK replays subscription state -// itself, so leave it alone. A full rejoin builds a new subscriber peer +// itself, so leave it alone. A reconnect builds a new subscriber peer // connection with no subscriptions and the SDK restores only what we publish, // so re-issue the subscriptions here. func (r *Room) onReconnected() { prev := r.reconnect.Swap(nil) - r.stats.Reconnecting.Store(false) + r.stats.Recovering.Store(false) room := r.room.Load() if room == nil { return } - // The SID is stable across a resume and changes on a full rejoin. Treat an - // unknown previous SID as a rejoin: re-subscribing is idempotent, while + // The SID is stable across a resume and changes on a reconnect. Treat an + // unknown previous SID as a reconnect: re-subscribing is idempotent, while // missing one leaves the call with no inbound room audio. sid := room.LocalParticipant.SID() - rejoined := prev == nil || prev.sid != sid + resumed := prev != nil && prev.sid == sid var gap time.Duration if prev != nil { gap = time.Since(prev.startedAt) } - if rejoined { - r.stats.Rejoins.Add(1) + if resumed { + r.stats.Resumes.Add(1) + } else { + r.stats.Reconnects.Add(1) } r.setParticipantFromRoom() - r.roomLog.Infow("reconnected to room", - "rejoined", rejoined, + r.roomLog.Infow("recovered connection to room", + "kind", recoveryKind(resumed), "gap", gap, "participantID", sid, "previousParticipantID", func() string { @@ -611,7 +621,7 @@ func (r *Room) onReconnected() { }(), ) - if !rejoined { + if resumed { return } if !r.subscribe.Load() { @@ -619,17 +629,17 @@ func (r *Room) onReconnected() { // into a leg that has not been accepted. return } - // The SDK calls this from the rejoin itself, inside the join's timeout, and - // subscribing does a blocking websocket write per track. Pass the room we + // The SDK calls this from the reconnect itself, inside the join's timeout, + // and subscribing does a blocking websocket write per track. Pass the room we // already read so a concurrent close cannot make this a nil dereference. - go r.resubscribeAfterRejoin(room) + go r.resubscribeAfterReconnect(room) } -func (r *Room) resubscribeAfterRejoin(room *lksdk.Room) { +func (r *Room) resubscribeAfterReconnect(room *lksdk.Room) { if r.closed.IsBroken() || r.stopped.IsBroken() { return } - r.roomLog.Infow("re-subscribing to remote tracks after rejoin") + r.roomLog.Infow("re-subscribing to remote tracks after reconnect") r.subscribeAll(room) } diff --git a/pkg/sip/room_test.go b/pkg/sip/room_test.go index 2ed6b1dc4..581b54b09 100644 --- a/pkg/sip/room_test.go +++ b/pkg/sip/room_test.go @@ -48,8 +48,8 @@ func testLocalInfo(sid string) *livekit.ParticipantInfo { } // testRemoteInfo is the other party in the room, holding one audio track. It -// never reconnects, so its SIDs stay the same across our rejoin, which is what -// makes the SDK treat its track as already known. +// never reconnects, so its SIDs stay the same across our reconnect, which is +// what makes the SDK treat its track as already known. func testRemoteInfo() []*livekit.ParticipantInfo { return []*livekit.ParticipantInfo{{ Sid: testRemoteSID, @@ -110,17 +110,17 @@ func (f *reconnectFixture) join(t *testing.T) { f.published.Store(0) // only count what happens from here on } -// rejoinEscalated simulates a failed resume escalating to a full rejoin, where +// reconnectEscalated simulates a failed resume escalating to a reconnect, where // the SDK skips OnRestarting. -func (f *reconnectFixture) rejoinEscalated() { +func (f *reconnectFixture) reconnectEscalated() { f.sdk.OnResuming() f.sdk.OnRoomJoined(testRoomInfo(), testLocalInfo("PA_sip_2"), testRemoteInfo(), &livekit.ServerInfo{}, nil) f.sdk.OnRestarted(testRoomInfo(), testLocalInfo("PA_sip_2"), testRemoteInfo()) } -// rejoinServerInitiated simulates the server asking for a full reconnect, where -// OnRestarting does run. -func (f *reconnectFixture) rejoinServerInitiated() { +// reconnectServerInitiated simulates the server asking for a reconnect directly, +// where OnRestarting does run. +func (f *reconnectFixture) reconnectServerInitiated() { f.sdk.OnRestarting() f.sdk.OnRoomJoined(testRoomInfo(), testLocalInfo("PA_sip_2"), testRemoteInfo(), &livekit.ServerInfo{}, nil) f.sdk.OnRestarted(testRoomInfo(), testLocalInfo("PA_sip_2"), testRemoteInfo()) @@ -136,47 +136,48 @@ func (f *reconnectFixture) rejoinServerInitiated() { // Only participants connected to the affected pod reconnect. Everyone other // participant in that room is unaffected and keep their session and their SIDs. // So "clearing participants" below means dropping our own view of them, not -// removing anyone from the room. There are two ways to reach a full rejoin: +// removing anyone from the room. There are two ways to reach a reconnect: // -// 1- server-initiated: rejoin is attempted right away, participants affected +// 1- server-initiated: the reconnect is attempted right away, affected participants // drop their participant map, and then rebuild it from the OnRoomJoined // snapshot and re-announce their tracks. // 2- resume-then-escalate: resume is first attempted. If it fails, then switch -// to a rejoin, and the stale participant map survives and no tracks are announced. +// to a reconnect, and the stale participant map survives so no tracks are +// announced. func TestRoomReconnect(t *testing.T) { - t.Run("escalated rejoin does not re-announce tracks from remote participants", func(t *testing.T) { + t.Run("escalated reconnect does not re-announce tracks from remote participants", func(t *testing.T) { f := newReconnectFixture(t) f.join(t) - f.rejoinEscalated() + f.reconnectEscalated() require.EqualValues(t, 0, f.published.Load(), "OnTrackPublished must not re-fire on the escalated path") }) // The other path does re-announce, which is why lost audio is intermittent. - t.Run("server initiated rejoin re-announces tracks from remote participants", func(t *testing.T) { + t.Run("server initiated reconnect re-announces tracks from remote participants", func(t *testing.T) { f := newReconnectFixture(t) f.join(t) - f.rejoinServerInitiated() + f.reconnectServerInitiated() require.EqualValues(t, 1, f.published.Load(), "OnTrackPublished is expected to re-fire when OnRestarting cleared the participants") }) - // However the rejoin was reached, SIP must re-issue its subscriptions. - t.Run("resubscribes after escalated rejoin", func(t *testing.T) { + // However the reconnect was reached, SIP must re-issue its subscriptions. + t.Run("resubscribes after escalated reconnect", func(t *testing.T) { f := newReconnectFixture(t) f.join(t) before := f.room.stats.TrackSubscribes.Load() - f.rejoinEscalated() + f.reconnectEscalated() require.Eventually(t, func() bool { return f.room.stats.TrackSubscribes.Load() > before }, time.Second, 10*time.Millisecond, - "SIP must re-subscribe to remote tracks after a full rejoin") + "SIP must re-subscribe to remote tracks after a reconnect") }) // An outbound call joins the room and publishes before it starts dialing, but @@ -190,7 +191,7 @@ func TestRoomReconnect(t *testing.T) { require.False(t, f.room.subscribe.Load()) before := f.room.stats.TrackSubscribes.Load() - f.rejoinEscalated() + f.reconnectEscalated() require.Never(t, func() bool { return f.room.stats.TrackSubscribes.Load() > before @@ -214,18 +215,20 @@ func TestRoomReconnect(t *testing.T) { "re-subscribing on resume would race the SDK's own sendSyncState") }) - t.Run("counts rejoins separately from resumes", func(t *testing.T) { + // The two counters are mutually exclusive, so a recovery lands in exactly one. + t.Run("counts resumes and reconnects separately", func(t *testing.T) { f := newReconnectFixture(t) f.join(t) - f.rejoinEscalated() - require.EqualValues(t, 1, f.room.stats.Rejoins.Load()) + f.reconnectEscalated() + require.EqualValues(t, 1, f.room.stats.Reconnects.Load()) + require.EqualValues(t, 0, f.room.stats.Resumes.Load()) f.sdk.OnResuming() f.sdk.OnResumed() - require.EqualValues(t, 2, f.room.stats.Reconnects.Load()) - require.EqualValues(t, 1, f.room.stats.Rejoins.Load(), "a resume must not count as a rejoin") - require.False(t, f.room.stats.Reconnecting.Load()) + require.EqualValues(t, 1, f.room.stats.Reconnects.Load(), "a resume must not count as a reconnect") + require.EqualValues(t, 1, f.room.stats.Resumes.Load()) + require.False(t, f.room.stats.Recovering.Load()) }) // PublishedFrames and PublishTX keep climbing during a gap whether or not @@ -235,11 +238,11 @@ func TestRoomReconnect(t *testing.T) { f.join(t) f.sdk.OnResuming() - require.True(t, f.room.stats.Load().Reconnecting, "gap must be visible while it is happening") + require.True(t, f.room.stats.Load().Recovering, "gap must be visible while it is happening") f.sdk.OnResumed() - require.False(t, f.room.stats.Load().Reconnecting) - require.EqualValues(t, 1, f.room.stats.Load().Reconnects) + require.False(t, f.room.stats.Load().Recovering) + require.EqualValues(t, 1, f.room.stats.Load().Resumes) }) // A SIP leg can hang up at any point, including mid-recovery, so teardown @@ -254,7 +257,7 @@ func TestRoomReconnect(t *testing.T) { wg.Add(2) go func() { defer wg.Done() - f.rejoinEscalated() + f.reconnectEscalated() }() go func() { defer wg.Done() @@ -265,18 +268,18 @@ func TestRoomReconnect(t *testing.T) { require.Nil(t, f.room.Room(), "close must clear the room handle") }) - // The SID changes on every full rejoin and feeds call state. - t.Run("refreshes participant SID after rejoin", func(t *testing.T) { + // The SID changes on every reconnect and feeds call state. + t.Run("refreshes participant SID after reconnect", func(t *testing.T) { f := newReconnectFixture(t) f.join(t) f.room.setParticipantFromRoom() require.Equal(t, "PA_sip_1", f.room.Participant().ID) - f.rejoinEscalated() + f.reconnectEscalated() require.Eventually(t, func() bool { return f.room.Participant().ID == "PA_sip_2" }, time.Second, 10*time.Millisecond, - "cached participant SID must be refreshed after a rejoin") + "cached participant SID must be refreshed after a reconnect") }) } From d803ef0e34573c66cc59b94661b4141591e3c892 Mon Sep 17 00:00:00 2001 From: Genseric Ghiro Date: Fri, 7 Aug 2026 15:13:50 -0400 Subject: [PATCH 5/5] Organizing imports --- pkg/sip/room.go | 1 + 1 file changed, 1 insertion(+) diff --git a/pkg/sip/room.go b/pkg/sip/room.go index 4b432967a..862a8641a 100644 --- a/pkg/sip/room.go +++ b/pkg/sip/room.go @@ -37,6 +37,7 @@ import ( "github.com/livekit/protocol/logger" "github.com/livekit/protocol/sip" lksdk "github.com/livekit/server-sdk-go/v2" + "github.com/livekit/sip/pkg/config" "github.com/livekit/sip/pkg/media/opus" )