mirror of
https://github.com/livekit/livekit.git
synced 2026-11-12 20:49:07 +00:00
rtc: don't let the forwarded RPC timeout underflow (#5154)
* rtc: don't let the forwarded RPC timeout underflow PerformRpc forwarded "responseTimeout - p.lastRTT" to the callee, both unsigned. A caller timeout below the callee's round trip, such as 100 ms against 150 ms, wrapped around to about 4.29 billion ms, so the callee was told it had roughly 49 days to respond while the server's own timer still used the real value. lastRTT was also read without p.lock, which UpdateMediaRTT holds when it writes it. Read lastRTT under the lock and keep the caller's timeout when the round trip is not below it. Fixes #5151 * rtc: compute the forwarded RPC timeout inline in PerformRpc Drop the rpcRequestTimeoutMs helper and test the forwarded ResponseTimeoutMs through PerformRpc instead.
This commit is contained in:
1 parent
45b5584c84
commit
401ddb1432
2 files changed
+56
-1
No files matched your search
+10
-1
@@ -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,
|
||||
},
|
||||
},
|
||||
|
||||
@@ -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())
|
||||
})
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user