diff --git a/pkg/sip/outbound_utilities_test.go b/pkg/sip/outbound_utilities_test.go index e678167ae..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", @@ -159,12 +160,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 +174,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..862a8641a 100644 --- a/pkg/sip/room.go +++ b/pkg/sip/room.go @@ -31,18 +31,21 @@ 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" ) +// 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"` @@ -65,6 +68,13 @@ type RoomStatsSnapshot struct { JitterBufferPacketsLost uint64 `json:"jitter_buffer_packets_lost"` JitterBufferPacketsDropped uint64 `json:"jitter_buffer_packets_dropped"` + TrackSubscribes uint64 `json:"track_subscribes"` + Resumes uint64 `json:"resumes"` + Reconnects uint64 `json:"reconnects"` + // Recovering reports whether the signal connection was down when the + // snapshot was taken. PublishedFrames and PublishTX are unreliable while set. + Recovering bool `json:"recovering"` + LatencyOutRecv LatencyStatsSnapshot `json:"latency_out_recv"` Closed bool `json:"closed"` @@ -78,6 +88,23 @@ type RoomStats struct { rtpStats rtpCountingStats dataPackets atomic.Uint64 + // TrackSubscribes counts subscribe requests issued for remote tracks. + // Attempts, not confirmations. + TrackSubscribes atomic.Uint64 + + // 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 + + // 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. + Recovering atomic.Bool + JitterBufferPacketsLost atomic.Uint64 JitterBufferPacketsDropped atomic.Uint64 @@ -110,6 +137,11 @@ func (s *RoomStats) Load() RoomStatsSnapshot { JitterBufferPacketsLost: s.JitterBufferPacketsLost.Load(), JitterBufferPacketsDropped: s.JitterBufferPacketsDropped.Load(), + TrackSubscribes: s.TrackSubscribes.Load(), + Resumes: s.Resumes.Load(), + Reconnects: s.Reconnects.Load(), + Recovering: s.Recovering.Load(), + PublishedFrames: s.PublishedFrames.Load(), PublishedSamples: s.PublishedSamples.Load(), PublishTX: math.Float64frombits(s.PublishTX.Load()), @@ -173,13 +205,17 @@ 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 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] + // 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] ready core.Fuse subscribe atomic.Bool subscribed core.Fuse @@ -224,8 +260,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() } @@ -250,10 +286,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 } - return r.room.DisconnectReason() + room := r.room.Load() + if room == nil { + return livekit.DisconnectReason_UNKNOWN_REASON + } + return room.DisconnectReason() } func (r *Room) Subscribed() <-chan struct{} { @@ -267,7 +307,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) { @@ -295,6 +335,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 +348,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.Store(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 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. +func (r *Room) setParticipantFromRoom() { + room := r.room.Load() + if room == nil { + return + } + p := ParticipantInfo{} + if cur := r.p.Load(); cur != nil { + p = *cur } - roomCallback := &lksdk.RoomCallback{ + 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,74 +532,139 @@ 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 reconnect and report the gap. +type reconnectState struct { + startedAt time.Time + sid string +} + +func recoveryKind(resumed bool) string { + if resumed { + return "resume" } - room := lksdk.NewRoom(roomCallback) - room.SetLogger(newRoomOverrideLogger(r.log)) - err := room.JoinWithContextAndToken(ctx, rconf.WsUrl, rconf.Token, - lksdk.WithAutoSubscribe(false), - lksdk.WithExtraAttributes(partConf.Attributes), + return "reconnect" +} + +// onReconnecting runs when the SDK loses the signal connection and starts +// 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() { + var sid string + if room := r.room.Load(); room != nil { + sid = room.LocalParticipant.SID() + } + r.reconnect.Store(&reconnectState{startedAt: time.Now(), sid: 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 reconnecting from scratch. +// +// A resume keeps the peer connections, and the SDK replays subscription state +// 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.Recovering.Store(false) + + room := r.room.Load() + if room == nil { + return + } + + // 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() + resumed := prev != nil && prev.sid == sid + + var gap time.Duration + if prev != nil { + gap = time.Since(prev.startedAt) + } + if resumed { + r.stats.Resumes.Add(1) + } else { + r.stats.Reconnects.Add(1) + } + + r.setParticipantFromRoom() + r.roomLog.Infow("recovered connection to room", + "kind", recoveryKind(resumed), + "gap", gap, + "participantID", sid, + "previousParticipantID", func() string { + if prev == nil { + return "" + } + return prev.sid + }(), ) - if err != nil { - return err + + if resumed { + 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 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.resubscribeAfterReconnect(room) +} - // Not subscribing to any tracks just yet! - return nil +func (r *Room) resubscribeAfterReconnect(room *lksdk.Room) { + if r.closed.IsBroken() || r.stopped.IsBroken() { + return + } + r.roomLog.Infow("re-subscribing to remote tracks after reconnect") + r.subscribeAll(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() { - if r.room == nil { + room := r.room.Load() + 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) @@ -556,9 +739,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() @@ -571,7 +753,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) { @@ -579,7 +764,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 { @@ -597,7 +786,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 { diff --git a/pkg/sip/room_test.go b/pkg/sip/room_test.go new file mode 100644 index 000000000..581b54b09 --- /dev/null +++ b/pkg/sip/room_test.go @@ -0,0 +1,285 @@ +// 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 reconnect, 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 +} + +// reconnectEscalated simulates a failed resume escalating to a reconnect, where +// the SDK skips OnRestarting. +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()) +} + +// 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()) +} + +// 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 reconnect: +// +// 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 reconnect, and the stale participant map survives so no tracks are +// announced. +func TestRoomReconnect(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.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 reconnect re-announces tracks from remote participants", func(t *testing.T) { + f := newReconnectFixture(t) + f.join(t) + + f.reconnectServerInitiated() + + require.EqualValues(t, 1, f.published.Load(), + "OnTrackPublished is expected to re-fire when OnRestarting cleared the participants") + }) + + // 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.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 reconnect") + }) + + // 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.reconnectEscalated() + + 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") + }) + + // 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.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, 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 + // 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().Recovering, "gap must be visible while it is happening") + + f.sdk.OnResumed() + 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 + // 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.reconnectEscalated() + }() + 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 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.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 reconnect") + }) +}