fix(rpc): give a SIP request 3s for a SIP instance to claim it

.changeset/sip-client-claim-timeout.md

@@ -0,0 +1,6 @@
+---
+"github.com/livekit/protocol": patch
+"@livekit/protocol": patch
+---
+
+Give a SIP request 3s for a SIP instance to claim it

rpc/sip_client.go

@@ -14,7 +14,17 @@
 
 package rpc
 
-import "github.com/livekit/psrpc"
+import (
+	"time"
+
+	"github.com/livekit/psrpc"
+)
+
+// sipClaimTimeout is how long a SIP request waits for a SIP instance to claim it. psrpc's 1s
+// default is too short for a claim published from a SIP pod whose Redis connections went cold
+// while it sat idle. The window only bounds the wait for an unclaimed request; nothing is sent
+// again, so a longer one cannot place a call twice.
+const sipClaimTimeout = 3 * time.Second
 
 type SIPClient interface {
 	SIPInternalClient
@@ -32,7 +42,8 @@ func NewSIPClientWithParams(params ClientParams) (SIPClient, error) {
 	if params.Bus == nil {
 		return nil, nil
 	}
-	opts := params.Options()
+	// First, so a caller's own selection timeout still wins.
+	opts := append([]psrpc.ClientOption{psrpc.WithClientSelectTimeout(sipClaimTimeout)}, params.Options()...)
 
 	internalClient, err := NewSIPInternalClient(params.Bus, opts...)
 	if err != nil {

rpc/sip_client_test.go

@@ -0,0 +1,76 @@
+package rpc
+
+import (
+	"context"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/livekit/psrpc"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/protobuf/proto"
+)
+
+// lateClaimBus holds back the claims a SIP instance publishes, the way a pod's first publish after
+// an idle stretch waits on a fresh Redis connection.
+type lateClaimBus struct {
+	psrpc.MessageBus
+	delay time.Duration
+}
+
+func (b lateClaimBus) Publish(ctx context.Context, channel psrpc.Channel, msg proto.Message) error {
+	if strings.HasSuffix(channel.Legacy, "|CLAIM") {
+		time.Sleep(b.delay)
+	}
+	return b.MessageBus.Publish(ctx, channel, msg)
+}
+
+type answeringSIPServer struct{}
+
+func (answeringSIPServer) CreateSIPParticipant(context.Context, *InternalCreateSIPParticipantRequest) (*InternalCreateSIPParticipantResponse, error) {
+	return &InternalCreateSIPParticipantResponse{ParticipantId: "PA_late"}, nil
+}
+
+func (answeringSIPServer) TransferSIPParticipant(context.Context, *InternalTransferSIPParticipantRequest) (*InternalTransferSIPParticipantResponse, error) {
+	return &InternalTransferSIPParticipantResponse{}, nil
+}
+
+func TestSIPClientWaitsForALateClaim(t *testing.T) {
+	const claimDelay = 1500 * time.Millisecond
+	require.Greater(t, claimDelay, psrpc.DefaultAffinityTimeout)
+	require.Less(t, claimDelay, sipClaimTimeout)
+
+	bus := psrpc.NewLocalMessageBus()
+	srv, err := NewSIPInternalServer(answeringSIPServer{}, lateClaimBus{bus, claimDelay})
+	require.NoError(t, err)
+	t.Cleanup(srv.Kill)
+	require.NoError(t, srv.RegisterCreateSIPParticipantTopic(""))
+
+	t.Run("psrpc's default window gives up first", func(t *testing.T) {
+		client, err := NewSIPInternalClient(bus)
+		require.NoError(t, err)
+		_, err = client.CreateSIPParticipant(context.Background(), "", &InternalCreateSIPParticipantRequest{})
+		require.ErrorIs(t, err, psrpc.ErrNoResponse)
+	})
+
+	t.Run("the SIP client waits for the claim", func(t *testing.T) {
+		client, err := NewSIPClient(bus)
+		require.NoError(t, err)
+		// livekit-server passes the dial's own timeout, so only the claim window limits the call.
+		// Under psrpc's 3s request default the claim window and the whole request end together.
+		res, err := client.CreateSIPParticipant(context.Background(), "", &InternalCreateSIPParticipantRequest{},
+			psrpc.WithRequestTimeout(time.Minute))
+		require.NoError(t, err)
+		require.Equal(t, "PA_late", res.ParticipantId)
+	})
+
+	t.Run("a caller's own window still wins", func(t *testing.T) {
+		client, err := NewSIPClientWithParams(ClientParams{
+			Bus:           bus,
+			ClientOptions: []psrpc.ClientOption{psrpc.WithClientSelectTimeout(psrpc.DefaultAffinityTimeout)},
+		})
+		require.NoError(t, err)
+		_, err = client.CreateSIPParticipant(context.Background(), "", &InternalCreateSIPParticipantRequest{})
+		require.ErrorIs(t, err, psrpc.ErrNoResponse)
+	})
+}