diff --git a/pkg/service/roomservice.go b/pkg/service/roomservice.go index 2cdf373a2..0a5ce6b29 100644 --- a/pkg/service/roomservice.go +++ b/pkg/service/roomservice.go @@ -18,6 +18,7 @@ import ( "context" "fmt" "strconv" + "time" "github.com/twitchtv/twirp" @@ -27,6 +28,7 @@ import ( "github.com/livekit/protocol/livekit" "github.com/livekit/protocol/logger" "github.com/livekit/protocol/rpc" + "github.com/livekit/protocol/utils" "github.com/livekit/psrpc" ) @@ -406,7 +408,22 @@ func (s *RoomService) PerformRpc(ctx context.Context, req *livekit.PerformRpcReq return nil, ErrDestinationIdentityRequired } - res, err := s.participantClient.PerformRpc(ctx, s.topicFormatter.ParticipantTopic(ctx, roomName, livekit.ParticipantIdentity(req.DestinationIdentity)), req) + res, err := s.participantClient.PerformRpc( + ctx, + s.topicFormatter.ParticipantTopic(ctx, roomName, livekit.ParticipantIdentity(req.DestinationIdentity)), + req, + psrpc.WithRequestTimeout(performRpcRequestTimeout(req)), + ) RecordResponse(ctx, res) return res, err } + +// performRpcRequestTimeout covers the participant's response window plus the ack round trip, +// since the psrpc default request timeout is shorter than the default RPC response timeout +func performRpcRequestTimeout(req *livekit.PerformRpcRequest) time.Duration { + responseTimeout := time.Duration(req.GetResponseTimeoutMs()) * time.Millisecond + if responseTimeout <= 0 { + responseTimeout = utils.DataChannelRpcDefaultResponseTimeout + } + return responseTimeout + utils.DataChannelRpcMaxRoundTripLatency +} diff --git a/pkg/service/roomservice_test.go b/pkg/service/roomservice_test.go index 5b27f2fdf..dc2575fb9 100644 --- a/pkg/service/roomservice_test.go +++ b/pkg/service/roomservice_test.go @@ -17,6 +17,7 @@ package service_test import ( "context" "testing" + "time" "github.com/stretchr/testify/require" "github.com/twitchtv/twirp" @@ -25,6 +26,8 @@ import ( "github.com/livekit/protocol/livekit" "github.com/livekit/protocol/rpc" "github.com/livekit/protocol/rpc/rpcfakes" + "github.com/livekit/protocol/utils" + "github.com/livekit/psrpc" "github.com/livekit/livekit-server/pkg/config" "github.com/livekit/livekit-server/pkg/routing/routingfakes" @@ -179,6 +182,7 @@ func newTestRoomService(limitConf config.LimitConfig) *TestRoomService { router := &routingfakes.FakeRouter{} allocator := &servicefakes.FakeRoomAllocator{} store := &servicefakes.FakeServiceStore{} + participantClient := &rpcfakes.FakeTypedParticipantClient{} svc, err := service.NewRoomService( limitConf, config.APIConfig{ExecutionTimeout: 2}, @@ -188,22 +192,55 @@ func newTestRoomService(limitConf config.LimitConfig) *TestRoomService { nil, rpc.NewTopicFormatter(), &rpcfakes.FakeTypedRoomClient{}, - &rpcfakes.FakeTypedParticipantClient{}, + participantClient, ) if err != nil { panic(err) } return &TestRoomService{ - RoomService: *svc, - router: router, - allocator: allocator, - store: store, + RoomService: *svc, + router: router, + allocator: allocator, + store: store, + participantClient: participantClient, } } +func TestPerformRpcRequestTimeout(t *testing.T) { + grant := &auth.ClaimGrants{ + Video: &auth.VideoGrant{RoomAdmin: true, Room: "testroom"}, + } + ctx := service.WithGrants(context.Background(), grant, "") + + requestTimeout := func(t *testing.T, svc *TestRoomService, call int) time.Duration { + t.Helper() + _, _, _, opts := svc.participantClient.PerformRpcArgsForCall(call) + var ro psrpc.RequestOpts + for _, o := range opts { + o(&ro) + } + return ro.Timeout + } + + t.Run("default response timeout", func(t *testing.T) { + svc := newTestRoomService(config.LimitConfig{}) + _, err := svc.PerformRpc(ctx, &livekit.PerformRpcRequest{Room: "testroom", DestinationIdentity: "p", Method: "m"}) + require.NoError(t, err) + require.Equal(t, utils.DataChannelRpcDefaultResponseTimeout+utils.DataChannelRpcMaxRoundTripLatency, requestTimeout(t, svc, 0)) + }) + + t.Run("requested response timeout", func(t *testing.T) { + svc := newTestRoomService(config.LimitConfig{}) + _, err := svc.PerformRpc(ctx, &livekit.PerformRpcRequest{Room: "testroom", DestinationIdentity: "p", Method: "m", ResponseTimeoutMs: 500}) + require.NoError(t, err) + require.Equal(t, 500*time.Millisecond+utils.DataChannelRpcMaxRoundTripLatency, requestTimeout(t, svc, 0)) + }) +} + type TestRoomService struct { service.RoomService - router *routingfakes.FakeRouter - allocator *servicefakes.FakeRoomAllocator - store *servicefakes.FakeServiceStore + router *routingfakes.FakeRouter + allocator *servicefakes.FakeRoomAllocator + store *servicefakes.FakeServiceStore + participantClient *rpcfakes.FakeTypedParticipantClient }