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