Port data streams over to rust livekit-ffi version - #769
Conversation
Replaces the hand-rolled Python data streams (client-side header/chunk/trailer framing over SendStreamHeader/Chunk/TrailerRequest) with the FFI's native implementation (data streams v2: single-packet streams + deflate): - Writers open handle-based FFI streams (text/byte_stream_open/write/close); chunking, stream ids, and compression now happen in rust - send_text/send_file are one-shot FFI calls and gain a compress option - Incoming streams arrive via text/byte_stream_opened room events; readers consume per-handle reader events after a read_incremental request, and unhandled topics dispose the owned reader handle - New public API: StreamError (raised when a stream terminates abnormally, previously indistinguishable from clean EOS) and RoomOptions.data_stream/DataStreamOptions.max_payload_byte_length - Drops the use_legacy_client_implementation override, so data streams v2 support is advertised to other clients The consumer-facing API is unchanged. Behavior changes: read errors now raise StreamError, stream ids are rust-generated unless supplied, sender_identity is honored consistently (was ignored for chunks/trailers), and trailers of targeted streams are no longer broadcast room-wide.
74bc9b2 to
fd370b3
Compare
Use a less hard to reason about PRNG randomness source
| """ | ||
| End-to-end tests for data streams (text/byte streams over the FFI-backed | ||
| rust implementation). Mirrors the node SDK's e2e_data_streams tests, plus | ||
| coverage of receiver-side payload caps, abnormal termination, and trailer | ||
| attributes. | ||
|
|
||
| Requirements: | ||
| - LIVEKIT_URL: LiveKit server URL | ||
| - LIVEKIT_API_KEY: API key for authentication | ||
| - LIVEKIT_API_SECRET: API secret for authentication | ||
|
|
||
| Tests will be skipped if these environment variables are not set. | ||
|
|
||
| Usage: | ||
| pytest test_e2e_data_streams.py -v | ||
| """ |
There was a problem hiding this comment.
I had a LLM generate these based on the node sdk data streams v2 e2e tests. I still need to review them in depth but with a cursory glance they look good.
![devin-ai-integration[bot]](https://avatars.githubusercontent.com/in/811515?s=60&v=4){kind=small}
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
… an error reading via ffi
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
…ch is closed in the background
size should be 0 like the current `main` state when a data stream does not have an associated size
|
I need to do another round of testing on this next week. There's been a bit more churn than I expected due to devin comments. |
| for ffi_handle in list(self._open_stream_writers.values()): | ||
| try: | ||
| ffi_handle.dispose() | ||
| except Exception: | ||
| logger.exception("failed to dispose data stream writer handle") | ||
| self._open_stream_writers.clear() |
There was a problem hiding this comment.
🟡 A data stream left open when the room disconnects still looks usable, so later writes or closes silently do nothing
The underlying resource for a still-open outgoing data stream is released (ffi_handle.dispose() at livekit-rtc/livekit/rtc/participant.py:281) without marking the stream itself as finished, so a later write or close on that stream targets something that no longer exists.
Impact: Code that finishes sending a file or message after the room has gone away can appear to succeed while the data is never sent, or fail with a confusing internal error instead of a clear "stream closed" message.
Handle released at disconnect while the writer object keeps _closed=False and a stale handle id
Room.disconnect() (livekit-rtc/livekit/rtc/room.py:689) and the end of _listen_task (livekit-rtc/livekit/rtc/room.py:739) both call Room._dispose_open_stream_writers(), which forwards to LocalParticipant._dispose_open_stream_writers() and drops every registered writer handle natively.
The room only holds FfiHandle objects (weakly) — it has no reference to the TextStreamWriter/ByteStreamWriter Python objects. So after disposal the writer still has self._closed is False and self._writer_handle set to the now-released id (livekit-rtc/livekit/rtc/data_stream.py:404-407, 423-425).
Consequently await writer.write(...) (livekit-rtc/livekit/rtc/data_stream.py:532-540) and await writer.aclose() (livekit-rtc/livekit/rtc/data_stream.py:456-468) pass their _writer_handle guards and issue text_stream_write / text_stream_close requests against a released handle.
This is made worse by _wait_for_callback (livekit-rtc/livekit/rtc/data_stream.py:439-454): if the FFI cannot service the request the response oneof is unset, so async_id evaluates to 0, and the predicate getattr(e, callback_field).async_id == 0 matches the first unrelated event that arrives (the unset callback sub-message also has async_id == 0). HasField("error") is then False, so the failed write/close returns as if it succeeded.
A cheap fix would be for the disconnect cleanup to also mark the owning writers closed (e.g. register the writer, or a small callback, alongside the handle) so write/aclose raise a clear RuntimeError/StreamError instead of talking to a released handle.
Prompt for agents
At disconnect, Room._dispose_open_stream_writers() -> LocalParticipant._dispose_open_stream_writers() drops the native handles of all still-open data stream writers. However the registry (LocalParticipant._open_stream_writers) only maps handle id -> FfiHandle, so the corresponding TextStreamWriter/ByteStreamWriter Python objects are never told their handle is gone: they keep _closed = False and _writer_handle set to the released id.
As a result, application code that calls await writer.write(...) or await writer.aclose() after the room has disconnected passes the writer's own guards and issues an FFI request against a released handle. Combined with BaseStreamWriter._wait_for_callback in livekit-rtc/livekit/rtc/data_stream.py, where a failed request yields an unset response oneof and therefore async_id == 0, the predicate matches the first arbitrary event whose (unset) callback sub-message also has async_id == 0, so the failed operation can return as if it succeeded.
Consider making the disconnect cleanup able to invalidate the writer objects themselves (for example register a weak reference to the writer, or a small invalidation callback, alongside the FfiHandle) so that write()/aclose() after disconnect raise a clear error. Separately, _wait_for_callback should treat an async_id of 0 (i.e. the expected response field not being set) as a failure rather than waiting for / matching an arbitrary event.
Was this helpful? React with 👍 or 👎 to provide feedback.
1 participant
(The python version of livekit/node-sdks#697)
Previously, the python 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 python 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
maxPayloadByteLengthoption on theroom.connectcall:3. Throwing data stream errors rather than exclusively logging them
The old python 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. I've also introduced a new
StreamErroras a new error type for these newly thrown errors.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