Adding integration test for handling reconnections

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)
+	}
+}

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)
+	}
+}

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,