diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 4de579399..d8d1b965d 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -4780,6 +4780,15 @@ func (p *ParticipantImpl) PerformRpc(req *livekit.PerformRpcRequest, resultCh ch } }) + p.lock.RLock() + rtt := p.lastRTT + p.lock.RUnlock() + + requestTimeout := responseTimeout + if rtt < requestTimeout { + requestTimeout -= rtt + } + rpcRequest := &livekit.DataPacket{ Kind: livekit.DataPacket_RELIABLE, ParticipantIdentity: id, @@ -4788,7 +4797,7 @@ func (p *ParticipantImpl) PerformRpc(req *livekit.PerformRpcRequest, resultCh ch Id: id, Method: req.GetMethod(), Payload: req.GetPayload(), - ResponseTimeoutMs: responseTimeout - p.lastRTT, + ResponseTimeoutMs: requestTimeout, Version: 1, }, }, diff --git a/pkg/rtc/participant_internal_test.go b/pkg/rtc/participant_internal_test.go index 2e88c1c54..17d85c477 100644 --- a/pkg/rtc/participant_internal_test.go +++ b/pkg/rtc/participant_internal_test.go @@ -18,6 +18,7 @@ import ( "fmt" "regexp" "runtime" + "slices" "strings" "sync" "testing" @@ -1927,3 +1928,48 @@ func TestUpdateVideoTrackReachesQueuedReplacement(t *testing.T) { require.NoError(t, p.UpdateVideoTrack(&livekit.UpdateLocalVideoTrack{TrackSid: sid, Width: 1280, Height: 720})) require.EqualValues(t, 1280, p.pendingTracks[pendingTrackCid].trackInfos[0].Width) } + +// the response timeout forwarded to the callee is the caller's less the round trip, and must not +// wrap around when the round trip is not below the timeout. +func TestPerformRpcForwardedResponseTimeout(t *testing.T) { + cases := []struct { + name string + timeout uint32 + rtt uint32 + expected uint32 + }{ + {"no rtt", 15000, 0, 15000}, + {"rtt taken off", 15000, 150, 14850}, + {"rtt just under the timeout", 1000, 999, 1}, + {"rtt equal to the timeout", 1000, 1000, 1000}, + {"rtt above the timeout", 1000, 1500, 1000}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + p := newParticipantForTest("test") + p.lock.Lock() + p.lastRTT = c.rtt + p.lock.Unlock() + + errorCh := make(chan error, 2) + p.PerformRpc(&livekit.PerformRpcRequest{ + DestinationIdentity: "test", + Method: "method", + ResponseTimeoutMs: c.timeout, + }, make(chan string, 1), errorCh) + + // the reliable channel is not writable yet, so the request is held for when it opens + var held [][]byte + require.Eventually(t, func() bool { + p.reliableDataInfo.joiningMessageLock.Lock() + defer p.reliableDataInfo.joiningMessageLock.Unlock() + held = slices.Clone(p.reliableDataInfo.joiningUnsequencedMessages) + return len(held) == 1 + }, time.Second, time.Millisecond) + + dp := &livekit.DataPacket{} + require.NoError(t, proto.Unmarshal(held[0], dp)) + require.Equal(t, c.expected, dp.GetRpcRequest().GetResponseTimeoutMs()) + }) + } +}