From 23d1e9b1247554e5518cf2d7cc67f2f592877e48 Mon Sep 17 00:00:00 2001 From: Genseric Ghiro Date: Fri, 14 Aug 2026 15:17:05 -0400 Subject: [PATCH] Adding integration test for handling reconnections --- test/integration/proxy_test.go | 175 +++++++++++++++++++++++++++++ test/integration/reconnect_test.go | 162 ++++++++++++++++++++++++++ test/integration/sip_test.go | 9 +- 3 files changed, 345 insertions(+), 1 deletion(-) create mode 100644 test/integration/proxy_test.go create mode 100644 test/integration/reconnect_test.go diff --git a/test/integration/proxy_test.go b/test/integration/proxy_test.go new file mode 100644 index 00000000..e7751371 --- /dev/null +++ b/test/integration/proxy_test.go @@ -0,0 +1,175 @@ +package integration + +import ( + "bufio" + "fmt" + "io" + "net" + "net/url" + "os" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +var debugSignalProxy = os.Getenv("DEBUG_SIGNAL_PROXY") != "" + +// signalProxy sits in front of the LiveKit signal connection so a test can break +// it. Only the SIP service is pointed at it: media uses separate ports and the +// test's own LiveKit clients dial the server directly, so cutting the proxy takes +// out signaling and nothing else. +type signalProxy struct { + ln net.Listener + upstream string + + // blockResume rejects reconnects that ask the server to resume the existing + // session. The SDK escalates to a full reconnect once a resume fails, which + // is the path worth testing. + blockResume atomic.Bool + resumesBlocked atomic.Int32 + + mu sync.Mutex + conns map[net.Conn]struct{} +} + +func newSignalProxy(t testing.TB, upstreamWsURL string) *signalProxy { + u, err := url.Parse(upstreamWsURL) + require.NoError(t, err) + + // Resolve once, over IPv4. The harness hands us a "localhost:port" URL, and + // leaving that to be resolved per dial makes every reconnect race the + // dual-stack fallback. + upstream, err := net.ResolveTCPAddr("tcp4", u.Host) + require.NoError(t, err) + + ln, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + + p := &signalProxy{ + ln: ln, + upstream: upstream.String(), + conns: make(map[net.Conn]struct{}), + } + t.Cleanup(func() { + _ = ln.Close() + p.Cut() + }) + go p.serve() + + t.Log("Signal proxy:", p.URL(), "->", p.upstream) + return p +} + +// URL is what the SIP service should use as its ws_url. +func (p *signalProxy) URL() string { + return "ws://" + p.ln.Addr().String() +} + +// Cut drops every open connection the way an unplugged network would, and reports +// how many it dropped. A proxied session holds two, one to each side. +func (p *signalProxy) Cut() int { + p.mu.Lock() + conns := make([]net.Conn, 0, len(p.conns)) + for c := range p.conns { + conns = append(conns, c) + } + clear(p.conns) + p.mu.Unlock() + + for _, c := range conns { + _ = c.Close() + } + return len(conns) +} + +func (p *signalProxy) BlockResume(v bool) { + p.blockResume.Store(v) +} + +func (p *signalProxy) ResumesBlocked() int { + return int(p.resumesBlocked.Load()) +} + +func (p *signalProxy) Conns() int { + p.mu.Lock() + defer p.mu.Unlock() + return len(p.conns) +} + +// debugf traces connections when DEBUG_SIGNAL_PROXY is set. It writes to stderr +// rather than to testing.TB because connection goroutines can outlive the subtest +// that started them. +func (p *signalProxy) debugf(format string, args ...any) { + if !debugSignalProxy { + return + } + fmt.Fprintf(os.Stderr, "signal proxy %s: %s\n", + time.Now().Format("15:04:05.000"), fmt.Sprintf(format, args...)) +} + +func (p *signalProxy) serve() { + for { + c, err := p.ln.Accept() + if err != nil { + return + } + go p.handle(c) + } +} + +func (p *signalProxy) handle(client net.Conn) { + defer client.Close() + + // The websocket handshake is a plain HTTP request, and its first line carries + // the whole query string. That is enough to tell a resume from a fresh join. + br := bufio.NewReader(client) + reqLine, err := br.ReadString('\n') + if err != nil { + return + } + if p.blockResume.Load() && strings.Contains(reqLine, "reconnect=1") { + p.resumesBlocked.Add(1) + p.debugf("blocked resume: %.100q", reqLine) + return + } + + server, err := net.Dial("tcp", p.upstream) + if err != nil { + p.debugf("upstream dial failed: %v", err) + return + } + p.debugf("relaying: %.100q", reqLine) + defer server.Close() + + p.add(client, server) + defer p.remove(client, server) + + // The request line was consumed above, so replay it before relaying the rest. + if _, err = io.WriteString(server, reqLine); err != nil { + return + } + done := make(chan struct{}, 2) + go func() { _, _ = io.Copy(server, br); done <- struct{}{} }() + go func() { _, _ = io.Copy(client, server); done <- struct{}{} }() + <-done +} + +func (p *signalProxy) add(conns ...net.Conn) { + p.mu.Lock() + defer p.mu.Unlock() + for _, c := range conns { + p.conns[c] = struct{}{} + } +} + +func (p *signalProxy) remove(conns ...net.Conn) { + p.mu.Lock() + defer p.mu.Unlock() + for _, c := range conns { + delete(p.conns, c) + } +} diff --git a/test/integration/reconnect_test.go b/test/integration/reconnect_test.go new file mode 100644 index 00000000..ca62079c --- /dev/null +++ b/test/integration/reconnect_test.go @@ -0,0 +1,162 @@ +package integration + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "github.com/livekit/protocol/livekit" + + "github.com/livekit/sip/pkg/siptest" + "github.com/livekit/sip/test/lktest" +) + +const ( + reconnectAudioTimeout = 20 * time.Second + + // The SDK retries for about a minute before it gives up and ends the call, so + // wait past that. Failing earlier would just be impatience, not a defect. + reconnectStateTimeout = 75 * time.Second +) + +// TestSIPRoomReconnect covers what happens to a live call when the SIP service +// loses its signal connection to LiveKit and gets it back. +// +// Blocking the resume is what makes the interesting case reachable. The SDK always +// tries to resume first, and a resume restores subscriptions on its own. Only once +// a resume fails does it escalate to a full reconnect, where the subscriptions are +// ours to restore. +func TestSIPRoomReconnect(t *testing.T) { + lk := runLiveKit(t) + proxy := newSignalProxy(t, lk.WsUrl) + srv := runSIPServerWithWsURL(t, lk, proxy.URL()) + + t.Run("escalated reconnect", func(t *testing.T) { + const roomName = "test-reconnect" + room, cli, sid := startCall(t, lk, srv, roomName, "reconnect-cli", "+000000001") + + proxy.BlockResume(true) + require.NotZero(t, proxy.Cut(), "expected a live signal connection to cut") + + // Confirms the escalation was forced rather than the SDK happening to + // take a path we did not choose. + require.Eventually(t, func() bool { return proxy.ResumesBlocked() > 0 }, + 15*time.Second, 100*time.Millisecond, + "no resume was attempted, so the reconnect path was never exercised") + proxy.BlockResume(false) + + // A reconnect mints a new SID, a resume keeps it. + newSID := waitSIPParticipant(t, lk, roomName, func(cur string) bool { return cur != sid }, + reconnectStateTimeout, "SIP participant SID never changed, so this was not a full reconnect") + t.Log("SIP participant rejoined:", sid, "->", newSID) + + // Without re-subscribing, the call stays up with no inbound room audio. + requireAudio(t, room, cli, "after the reconnect") + }) + + // A resume keeps the session and the SDK replays subscriptions itself, so + // nothing on our side has to run. + t.Run("resume", func(t *testing.T) { + const roomName = "test-resume" + room, cli, sid := startCall(t, lk, srv, roomName, "resume-cli", "+000000002") + + blocked := proxy.ResumesBlocked() + require.NotZero(t, proxy.Cut(), "expected a live signal connection to cut") + + require.Eventually(t, func() bool { return proxy.Conns() > 0 }, + reconnectStateTimeout, 100*time.Millisecond, + "signal connection was never re-established") + + newSID := waitSIPParticipant(t, lk, roomName, nil, + reconnectStateTimeout, "SIP participant left the room") + require.Equal(t, sid, newSID, "SID changed, so the SDK did a full reconnect rather than a resume") + require.Equal(t, blocked, proxy.ResumesBlocked(), "proxy should not have blocked anything here") + + requireAudio(t, room, cli, "after the resume") + }) +} + +// startCall puts a LiveKit participant and a SIP caller in the same room and +// returns once audio flows between them. +func startCall(t *testing.T, lk *LiveKit, srv *SIPServer, roomName, clientID, trunkNumber string) (*lktest.Participant, *siptest.Client, string) { + t.Helper() + + nc := createTrunkAndDirect(t, srv, roomName, trunkNumber) + + room := lk.ConnectParticipant(t, roomName, "room-participant", nil) + cli := runClient(t, nc, srv.IP, clientID, clientNumber, false, nil, nil, nil, nil) + + // Keep the caller audible for the whole test. The SIP service drops a call + // whose media goes quiet, and an outage can last longer than that timeout. + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + go func() { _ = cli.SendSilence(ctx) }() + + sid := waitSIPParticipant(t, lk, roomName, nil, + participantsJoinTimeout, "SIP participant never joined the room") + requireAudio(t, room, cli, "before the outage") + return room, cli, sid +} + +// createTrunkAndDirect is CreateTrunkAndDirect with the dispatch rule scoped to +// its own trunk. The subtests share a LiveKit server, and two unscoped direct +// rules with no PIN collide. +func createTrunkAndDirect(t *testing.T, srv *SIPServer, roomName, trunkNumber string) *NumberConfig { + t.Helper() + + trunkID := srv.CreateTrunkIn(t, &livekit.SIPInboundTrunkInfo{ + Numbers: []string{trunkNumber}, + }) + dr, err := srv.Client.CreateSIPDispatchRule(context.Background(), &livekit.CreateSIPDispatchRuleRequest{ + Name: roomName, + TrunkIds: []string{trunkID}, + Rule: &livekit.SIPDispatchRule{ + Rule: &livekit.SIPDispatchRule_DispatchRuleDirect{ + DispatchRuleDirect: &livekit.SIPDispatchRuleDirect{RoomName: roomName}, + }, + }, + }) + require.NoError(t, err) + t.Log("New dispatch rule (direct):", dr.SipDispatchRuleId) + + return &NumberConfig{SIP: srv, TrunkID: trunkID, RuleID: dr.SipDispatchRuleId, Number: trunkNumber} +} + +// requireAudio checks that audio flows both ways between the room and the SIP leg. +// phase names the moment being checked, so a failure says which one. +func requireAudio(t *testing.T, room *lktest.Participant, cli *siptest.Client, phase string) { + t.Helper() + + ctx, cancel := context.WithTimeout(context.Background(), reconnectAudioTimeout) + defer cancel() + t.Log("checking audio", phase) + lktest.CheckAudioForParticipants(t, ctx, room, cli) +} + +// waitSIPParticipant polls until the room holds exactly one SIP participant whose +// SID satisfies cond, and returns that SID. A nil cond accepts any SID. A reconnect +// can briefly leave the previous participant behind, so wait for the room to settle +// on one. +func waitSIPParticipant(t *testing.T, lk *LiveKit, room string, cond func(sid string) bool, timeout time.Duration, msg string) string { + t.Helper() + + deadline := time.Now().Add(timeout) + for { + var sids []string + for _, p := range lk.RoomParticipants(t, room) { + if p.Kind == livekit.ParticipantInfo_SIP { + sids = append(sids, p.Sid) + } + } + if len(sids) == 1 && (cond == nil || cond(sids[0])) { + return sids[0] + } + if time.Now().After(deadline) { + t.Fatalf("%s (SIP participants: %v)", msg, sids) + return "" + } + time.Sleep(250 * time.Millisecond) + } +} diff --git a/test/integration/sip_test.go b/test/integration/sip_test.go index d6da07df..f4f0c70b 100644 --- a/test/integration/sip_test.go +++ b/test/integration/sip_test.go @@ -48,6 +48,13 @@ type SIPServer struct { } func runSIPServer(t testing.TB, lk *LiveKit) *SIPServer { + return runSIPServerWithWsURL(t, lk, lk.WsUrl) +} + +// runSIPServerWithWsURL routes the SIP service's room signal connection through +// wsURL. Nothing else in the harness reads it, so a test can point it at a proxy +// and break that one connection. +func runSIPServerWithWsURL(t testing.TB, lk *LiveKit, wsURL string) *SIPServer { rc, err := redis.GetRedisClient(lk.Redis) if err != nil { t.Fatal(err) @@ -64,7 +71,7 @@ func runSIPServer(t testing.TB, lk *LiveKit) *SIPServer { NodeID: utils.NewGuid("NS_"), ApiKey: lk.ApiKey, ApiSecret: lk.ApiSecret, - WsUrl: lk.WsUrl, + WsUrl: wsURL, Redis: lk.Redis, SIPPort: sipPort, SIPPortListen: sipPort,