From 3564cfbabe71ee412012ef68963723c0c7557641 Mon Sep 17 00:00:00 2001 From: Cosmin Staicu Date: Wed, 23 Sep 2026 19:25:08 +0300 Subject: [PATCH] fix(rpc): give a SIP request 3s for a SIP instance to claim it CreateSIPParticipant and TransferSIPParticipant are queue RPCs that need a claim: the caller publishes the request, one SIP instance takes the queue lock and publishes a claim back, and only then is the request granted. The SIP client never set a selection timeout, so psrpc's 1s default applied, and livekit-server builds this client without retries on purpose. A claim that lands after 1s fails the dial outright with "no response from servers". On the Redis bus that claim costs the SIP instance two round trips on pooled connections, the SetNX queue lock and the claim publish. After an idle stretch longer than the pool's idle limit both open fresh TLS connections first, which is when a dial has been seen to miss the window while the next one, moments later, goes through. NewSIPClientWithParams now puts a 3s selection timeout ahead of the caller's options, so a caller that sets its own still wins. The window only bounds how long an unclaimed request waits; the request is never sent again, so a longer one cannot place a call twice. A SIP instance that is really gone now fails in 3s instead of 1s. Signed-off-by: Cosmin Staicu --- .changeset/sip-client-claim-timeout.md | 6 ++ rpc/sip_client.go | 15 ++++- rpc/sip_client_test.go | 76 ++++++++++++++++++++++++++ 3 files changed, 95 insertions(+), 2 deletions(-) create mode 100644 .changeset/sip-client-claim-timeout.md create mode 100644 rpc/sip_client_test.go diff --git a/.changeset/sip-client-claim-timeout.md b/.changeset/sip-client-claim-timeout.md new file mode 100644 index 000000000..065931350 --- /dev/null +++ b/.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 diff --git a/rpc/sip_client.go b/rpc/sip_client.go index 8c1d20c72..095665fd6 100644 --- a/rpc/sip_client.go +++ b/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 { diff --git a/rpc/sip_client_test.go b/rpc/sip_client_test.go new file mode 100644 index 000000000..f8b357c33 --- /dev/null +++ b/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) + }) +}