From b9419f16cac6cf61bf393a42a05b4d31738dbdc8 Mon Sep 17 00:00:00 2001 From: David Zhao Date: Sat, 19 Sep 2026 19:10:31 -0700 Subject: [PATCH] improve flakey tests limit maxRoundTripLatency during tests --- .changeset/rpc-max-round-trip-latency.md | 5 ++ packages/livekit-rtc/src/participant.ts | 11 +++- packages/livekit-rtc/src/rpc.ts | 7 +++ packages/livekit-rtc/src/tests/e2e.test.ts | 67 ++++++++++------------ 4 files changed, 51 insertions(+), 39 deletions(-) create mode 100644 .changeset/rpc-max-round-trip-latency.md diff --git a/.changeset/rpc-max-round-trip-latency.md b/.changeset/rpc-max-round-trip-latency.md new file mode 100644 index 00000000..597e6b6f --- /dev/null +++ b/.changeset/rpc-max-round-trip-latency.md @@ -0,0 +1,5 @@ +--- +'@livekit/rtc-node': patch +--- + +`performRpc` accepts `maxRoundTripLatency`, the time to wait for the destination to acknowledge the request before failing with `CONNECTION_TIMEOUT` (default 7s). `RpcCallInfo` carries it so interceptors can adjust it. diff --git a/packages/livekit-rtc/src/participant.ts b/packages/livekit-rtc/src/participant.ts index 4a6bc974..07ed90a4 100644 --- a/packages/livekit-rtc/src/participant.ts +++ b/packages/livekit-rtc/src/participant.ts @@ -841,8 +841,15 @@ export class LocalParticipant extends Participant { method, payload, responseTimeout, + maxRoundTripLatency, }: PerformRpcParams): Promise { - const call: RpcCallInfo = { destinationIdentity, method, payload, responseTimeout }; + const call: RpcCallInfo = { + destinationIdentity, + method, + payload, + responseTimeout, + maxRoundTripLatency, + }; // snapshot the interceptor list so add/remove during a call is well defined const perform = chainOutgoing([...this.rpcInterceptors], (c) => this.performRpcFfi(c)); return await perform(call); @@ -853,6 +860,7 @@ export class LocalParticipant extends Participant { method, payload, responseTimeout, + maxRoundTripLatency, }: RpcCallInfo): Promise { const req = new PerformRpcRequest({ localParticipantHandle: this.ffi_handle.handle, @@ -860,6 +868,7 @@ export class LocalParticipant extends Participant { method, payload, responseTimeoutMs: responseTimeout, + maxRoundTripLatencyMs: maxRoundTripLatency, }); const res = FfiClient.instance.request({ diff --git a/packages/livekit-rtc/src/rpc.ts b/packages/livekit-rtc/src/rpc.ts index e0f5dedf..ebf7f7e3 100644 --- a/packages/livekit-rtc/src/rpc.ts +++ b/packages/livekit-rtc/src/rpc.ts @@ -13,6 +13,11 @@ export interface PerformRpcParams { payload: string; /** Timeout for receiving a response after initial connection (milliseconds). Default: 10000 */ responseTimeout?: number; + /** + * Maximum time to wait for the destination to acknowledge the request (milliseconds). + * If no ack arrives in time the call fails with `CONNECTION_TIMEOUT`. Default: 7000 + */ + maxRoundTripLatency?: number; } /** @@ -60,6 +65,8 @@ export interface RpcCallInfo { payload: string; /** Milliseconds to wait for a response, or `undefined` for the default. */ responseTimeout?: number; + /** Milliseconds to wait for the destination's ack, or `undefined` for the default. */ + maxRoundTripLatency?: number; } /** Continuation handed to {@link RpcInterceptor.interceptOutgoing}: performs the call. */ diff --git a/packages/livekit-rtc/src/tests/e2e.test.ts b/packages/livekit-rtc/src/tests/e2e.test.ts index 912feeef..b55e5bf6 100644 --- a/packages/livekit-rtc/src/tests/e2e.test.ts +++ b/packages/livekit-rtc/src/tests/e2e.test.ts @@ -30,6 +30,9 @@ import { // use concurrent testing if available on the runner (currently not supported by bun's api) const it = typeof itRaw.concurrent === 'function' ? itRaw.concurrent : itRaw; +// Test bodies take `expect` from the test context. With concurrent tests the module-level +// `expect` records `.resolves`/`.rejects` promises against whichever test vitest considers +// current, so one test's rejection gets reported as a failure of an unrelated test. /** * Only tracks published by `identity` are the test's business. Anything else in @@ -130,7 +133,7 @@ describeE2E('livekit-rtc e2e', () => { it( 'connects to a room', - async () => { + async ({ expect }) => { const { roomName, rooms } = await connectTestRooms(1); const room = rooms[0]!; @@ -153,7 +156,7 @@ describeE2E('livekit-rtc e2e', () => { it( 'connects multiple participants to the same room', - async () => { + async ({ expect }) => { const { roomName, rooms } = await connectTestRooms(2); const [first, second] = rooms; @@ -169,7 +172,7 @@ describeE2E('livekit-rtc e2e', () => { it( 'sends and receives a data message between participants sharing a key', - async () => { + async ({ expect }) => { // A shared-key `keyProviderOptions` with only `sharedKey` set relies on // the SDK filling in the remaining (proto-required) provider defaults; // regression guard for connect failing to encode KeyProviderOptions. @@ -209,7 +212,7 @@ describeE2E('livekit-rtc e2e', () => { it( 'emits participantDisconnected when a participant leaves', - async () => { + async ({ expect }) => { const { rooms } = await connectTestRooms(2); const [first, second] = rooms; const secondIdentity = second!.localParticipant!.identity; @@ -240,7 +243,7 @@ describeE2E('livekit-rtc e2e', () => { // transmitted silence, which the detector reads as a wrong frequency. itRaw( 'transfers audio between two participants (sine detection)', - async () => { + async ({ expect }) => { const cases = [ { pubRateHz: 48_000, pubChannels: 1, subRateHz: 48_000, subChannels: 1 }, { pubRateHz: 48_000, pubChannels: 2, subRateHz: 48_000, subChannels: 2 }, @@ -417,7 +420,7 @@ describeE2E('livekit-rtc e2e', () => { it( 'sends and receives text and byte streams', - async () => { + async ({ expect }) => { const { rooms } = await connectTestRooms(2); const [receivingRoom, sendingRoom] = rooms; const senderIdentity = sendingRoom!.localParticipant!.identity; @@ -481,7 +484,7 @@ describeE2E('livekit-rtc e2e', () => { it( 'invokes RPC methods and returns structured errors', - async () => { + async ({ expect }) => { const { rooms } = await connectTestRooms(2); const [callerRoom, calleeRoom] = rooms; @@ -511,25 +514,17 @@ describeE2E('livekit-rtc e2e', () => { }, }); - // `room.connect()` resolves on the signal handshake, so the first - // data-channel message still waits on ICE/DTLS/SCTP setup — seconds, on a - // small runner. Warm the channel up untimed so the assertions below - // measure RPC behavior rather than connection setup. - await callerRoom!.localParticipant!.performRpc({ - destinationIdentity: calleeRoom!.localParticipant!.identity, - method, - payload, - responseTimeout: testTimeoutMs, - }); - - const rpcResponseTimeoutMs = 1_000; - + // These calls assert on RPC semantics, not latency, so they get the full + // test budget. `room.connect()` resolves on the signal handshake, so the + // first data-channel message still waits on ICE/DTLS/SCTP setup, and the + // suite runs concurrently in one process on a small runner: a tight + // responseTimeout here was the most frequent flake on main. await expect( callerRoom!.localParticipant!.performRpc({ destinationIdentity: calleeRoom!.localParticipant!.identity, method, payload, - responseTimeout: rpcResponseTimeoutMs, + responseTimeout: testTimeoutMs, }), ).resolves.toBe(payload); @@ -538,35 +533,31 @@ describeE2E('livekit-rtc e2e', () => { destinationIdentity: calleeRoom!.localParticipant!.identity, method: 'unregistered-method', payload, - responseTimeout: rpcResponseTimeoutMs, + responseTimeout: testTimeoutMs, }), ).rejects.toMatchObject({ code: RpcError.ErrorCode.UNSUPPORTED_METHOD }); - expect(outgoing).toEqual([ - `${method}:${payload}->${payload}`, - `${method}:${payload}->${payload}`, - ]); + expect(outgoing).toEqual([`${method}:${payload}->${payload}`]); // the unregistered method never reached the callee's chain: the FFI layer rejects a // method nobody registered before the SDK's handler is invoked - expect(incoming).toEqual([ - `${method}:${callerRoom!.localParticipant!.identity}`, - `${method}:${callerRoom!.localParticipant!.identity}`, - ]); + expect(incoming).toEqual([`${method}:${callerRoom!.localParticipant!.identity}`]); - // Short by design: no ack ever arrives for an absent participant, so the - // timeout expiring *is* the behavior under test. + // No ack ever arrives for an absent participant, so the ack budget + // expiring *is* the behavior under test. Bound it explicitly: the FFI's + // default is 7s, which used to be most of this test's runtime. await expect( callerRoom!.localParticipant!.performRpc({ destinationIdentity: 'unknown-participant', method, payload, - responseTimeout: 500, + responseTimeout: testTimeoutMs, + maxRoundTripLatency: 500, }), ).rejects.toMatchObject({ code: RpcError.ErrorCode.CONNECTION_TIMEOUT }); await Promise.all(rooms.map((r) => r.disconnect())); }, - testTimeoutMs * 2, + testTimeoutMs, ); it( @@ -611,7 +602,7 @@ describeE2E('livekit-rtc e2e', () => { it( 'cleans up track publications when a remote participant disconnects', - async () => { + async ({ expect }) => { const { rooms } = await connectTestRooms(2); const [stayingRoom, leavingRoom] = rooms; @@ -664,7 +655,7 @@ describeE2E('livekit-rtc e2e', () => { it( 'cleans up resources when multiple participants disconnect simultaneously', - async () => { + async ({ expect }) => { // Connect 4 participants to stress-test concurrent disconnection cleanup const { rooms } = await connectTestRooms(4); @@ -719,7 +710,7 @@ describeE2E('livekit-rtc e2e', () => { it( 'concurrent getSid() calls share a single listener and resolve consistently', - async () => { + async ({ expect }) => { const { rooms } = await connectTestRooms(1); const room = rooms[0]!; @@ -883,7 +874,7 @@ describeE2E('livekit-rtc e2e', () => { itRaw( 'full reconnect keeps audio flowing and ends with one publication on the subscriber', - async () => { + async ({ expect }) => { const { rooms, subRoom, pubRoom } = await runReconnectScenario( SimulateScenarioKind.SIMULATE_FULL_RECONNECT, );