Port data streams over to rust livekit-ffi version - #697
Conversation
🦋 Changeset detectedLatest commit: a32e2e6 The changes in this PR will be included in the next version bump. This PR includes changesets to release 1 package
Not sure what this means? Click here to learn what changesets are. Click here if you're a maintainer who wants to add another changeset to this PR |
5f3717a to
3b9d64a
Compare
![devin-ai-integration[bot]](https://avatars.githubusercontent.com/in/811515?s=60&v=4){kind=small}
This comment was marked as outdated.
This comment was marked as outdated.
Sorry, something went wrong.
…m and not dropped Previously they were just being ignored, which meant that a "data stream too big" error (probably along with others too) would not get surfaced.
The data streams port left four waitFor predicates wrapped in a way prettier reformats, which failed the CI format:check gate. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
b6170e2 to
02cf89c
Compare
02cf89c to
d93b835
Compare
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
Co-authored-by: devin-ai-integration[bot] <158243242+devin-ai-integration[bot]@users.noreply.github.com>
| // SPDX-FileCopyrightText: 2026 LiveKit, Inc. | ||
| // | ||
| // SPDX-License-Identifier: Apache-2.0 | ||
| // | ||
| // End-to-end tests for data streams (text/byte streams over the FFI-backed | ||
| // rust implementation). Ported from rust-sdks `livekit/tests/data_stream_test.rs` | ||
| // where the node API surface allows; see comments on individual tests for | ||
| // intentional deviations. | ||
| import { afterAll, expect, it as itRaw } from 'vitest'; | ||
| import { dispose } from '../index.js'; | ||
| import { |
There was a problem hiding this comment.
I had a LLM generate these based on the rust sdk's e2e tests for data streams v2. I still need to review them in depth but with a cursory glance they look good.
| // Progress is reported against the payload's total byte length, so it has | ||
| // to be measured in received bytes too — a chunk count would report e.g. | ||
| // 1/50000 for a payload that arrived as a single chunk. | ||
| super.handleChunkReceived(chunk); |
There was a problem hiding this comment.
🟡 Long incoming text streams hold the entire message in memory until they finish
Every piece of an incoming text stream is kept in an internal lookup table (this.receivedChunks.set(index, chunk) at packages/livekit-rtc/src/data_streams/stream_reader.ts:115) even after it has already been handed to the caller, so memory keeps growing with the whole message instead of staying flat while it is consumed piece by piece.
Impact: An application that reads a large or long-running text stream incrementally still accumulates the full payload in memory, which can exhaust memory now that incoming streams may be up to 5 GB.
Why the retained map can never be pruned under the new receive path
TextStreamReader.handleChunkReceived stores each chunk keyed by chunkIndex purely so it can drop an older version of the same index (packages/livekit-rtc/src/data_streams/stream_reader.ts:108-116). The new FFI-backed receive path synthesizes chunks itself with only content and a monotonically increasing chunkIndex and never sets version (packages/livekit-rtc/src/room.ts:1050-1052, and the byte path at packages/livekit-rtc/src/room.ts:984-986), so the de-duplication branch is now dead and the map is pure retention. Entries are only released when the iterator observes done or an error (packages/livekit-rtc/src/data_streams/stream_reader.ts:141, 154), i.e. at the very end of the stream. Since v2 no longer re-sends versioned chunks, the map (and the bigIntToNumber(chunk.chunkIndex!) bookkeeping around it) can be dropped entirely, leaving only the byte-progress accounting from the base class.
Prompt for agents
In packages/livekit-rtc/src/data_streams/stream_reader.ts, TextStreamReader keeps a `receivedChunks` Map of every chunk it has seen, solely to drop older `version`s of the same `chunkIndex`. With the port to the FFI data streams v2 receive path, Room synthesizes DataStream_Chunk objects with only `content` and an incrementing `chunkIndex` and never populates `version` (see handleTextStreamOpened in packages/livekit-rtc/src/room.ts), so the de-duplication branch can never fire and the map simply retains the entire payload until the stream ends (it is only cleared on done/error/return in the async iterator). Consider removing the `receivedChunks` map and the version-based de-duplication from TextStreamReader so it relies on the base class byte-progress accounting only, or otherwise avoid retaining chunk contents that have already been yielded to the consumer.
Was this helpful? React with 👍 or 👎 to provide feedback.
1 participant
Previously, the node sdk had its own data streams implementation. My understanding is there isn't really a good reason for this, and it's more of an artifact of history - the node/python sdks had data streams added first, and the rust implementation came later on.
With data streams v2, the rust implementation will both be quite a bit more stable and gain some new features that make it significantly more performant (single packet data streams and DEFLATE compression when it makes the payload smaller). This roughly doubles data stream throughput in local testing.
So, port the node sdk to use the rust sdk data streams v2 implementation, and completely remove the typescript implementation. This is being done via the already-existing
livekit-ffidata streams messages which the C++ and unity sdks use today. This is a substantial change which needs thorough testing.New behaviors worth being aware of
All of these are either data streams v2 related changes, or bug fixes.
1.
compressoptionWhen sending a data stream, there is a new
compressoption. Just like how this works on web / rust, this defaults totrue. If set tofalse, then compression will be disabled (useful if you know the data you are sending isn't compressible / you are doing your own compression, which is not uncommon in robotics use cases). The vast majority of users should leave this set totrue.2. Max data stream size
In a rust data streams v2 pull request review comment, we decided that for security reasons it made sense to introduce a maximum data stream size as a DOS protection. This limit is by default
5gb- any data stream that is larger will now read up until that point, and if the stream keeps going, a "payload too large" error will be raised on the stream and exposed to a user on the subsequent.read()call.If a user is sending a large file, they can override this by setting a new
dataStream.maxPayloadByteLengthoption on theroom.connectcall:3. Throwing data stream errors rather than exclusively logging them
The old node specific data streams implementation didn't properly surface errors to the caller which were encountered while reading the stream. As a prerequisite for exposing the error for item 2, I fixed this, but it means that now if a data stream were to fail, errors would get thrown when in the past they would get logged and dropped.
Testing
I've tested this as best as I can locally but I'd really like some help from agents folks to really put this through the ringer and make sure I haven't inadvertently broken something.
Todo