Data streams v2 uniffi - #1286
Conversation
29ffd71 to
7ef488b
Compare
Changeset ✓This PR includes a changeset covering all affected packages:
|
![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.
|
My suggested approach here is to just open a draft PR to consumers e.g. swift and then polish it - that's what I did for data tracks before. Otherwise the reviewer must generate the bindings anyway, as it's very hard to predict the edge cases around concurrency, memory management, etc. I'd be super happy to take a look at both in parallel. |
f28049b to
d1c8eff
Compare
| /// Data streams v2 core from [`livekit-data-stream`]. | ||
| pub mod data_stream; |
There was a problem hiding this comment.
🟡 Pull request is missing the required changeset file
No changeset file was added for this change (new module registered at livekit-uniffi/src/lib.rs:19), even though the repository requires every pull request to include one listing the crates that need a version bump.
Impact: Release tooling will not record or version this new functionality.
Repository rule
AGENTS.md ("Documenting changes"): "Every PR needs a changeset" and "Changeset must list any crates which need to be bumped stemming from the change". The diff adds livekit-uniffi/src/data_stream/* and a new dependency in livekit-uniffi/Cargo.toml, but .changeset/ contains only pre-existing entries (fix-publisher-renegotiation-deadlock.md, fix_nvenc_dynamic_bitrate_updates.md, fix_uniffi_android_package_build.md).
Add a changeset (e.g. via knope document-change) listing livekit-uniffi.
Was this helpful? React with 👍 or 👎 to provide feedback.
| /// Handles an encoded [`livekit_protocol::DataPacket`] received over the data channel. | ||
| /// | ||
| /// Fire-and-forget: the packet is decoded and enqueued in order; processing happens on the | ||
| /// manager's run loop. Non-data-stream or undecodable packets are ignored. | ||
| pub fn handle_packet_received(&self, packet: Bytes) { | ||
| if let Some(event) = decode_data_packet(&packet) { | ||
| let _ = self.input.send(event.into()); | ||
| } | ||
| } |
There was a problem hiding this comment.
🟡 Readers can wait forever when the sending participant leaves mid-stream
There is no way for the host app to tell the receiver that a participant has left (no equivalent of the abort input used elsewhere, only handle_packet_received at livekit-uniffi/src/data_stream/incoming.rs:77), so a half-received stream is never ended and a pending read never returns.
Impact: If a sender disconnects in the middle of a stream, an app awaiting the stream contents hangs indefinitely and the partially received stream is retained.
The Rust room implementation aborts streams on disconnect; the FFI has no such entry point
The incoming actor supports ds::incoming::InputEvent::AbortStreamsFrom(identity), which the livekit crate sends when a remote participant disconnects (livekit/src/room/mod.rs:2269), causing the descriptor to be dropped and the reader's channel to error/close.
IncomingDataStreamManager only exposes handle_packet_received, so a foreign host cannot deliver that event. Consequently ByteStreamReader::read_all/next (livekit-uniffi/src/data_stream/incoming.rs:105-117) and the text equivalents (:141-153) await a chunk that will never arrive, while holding the reader's tokio mutex, and the manager keeps the open-stream descriptor alive.
Fix: expose a method such as handle_participant_disconnected(identity: String) that sends InputEvent::AbortStreamsFrom.
Prompt for agents
IncomingDataStreamManager in livekit-uniffi/src/data_stream/incoming.rs only exposes handle_packet_received, but the underlying incoming actor also accepts ds::incoming::InputEvent::AbortStreamsFrom(ParticipantIdentity), which the livekit crate sends on remote participant disconnect (see livekit/src/room/mod.rs:2269). Without an FFI entry point for it, foreign hosts cannot terminate in-flight streams whose sender left, leaving readers awaiting chunks forever and descriptors retained in the manager. Add a synchronous exported method that forwards an identity as InputEvent::AbortStreamsFrom, documented as required on participant disconnect.
Was this helpful? React with 👍 or 👎 to provide feedback.
I think this is going to be a lot cleaner, and mean that other platforms like swift can handle "internal" data streams in their own way
| #[uniffi::constructor] | ||
| pub fn new( | ||
| delegate: Arc<dyn IncomingDataStreamManagerDelegate>, | ||
| max_payload_byte_length: Option<u64>, | ||
| ) -> Arc<Self> { |
There was a problem hiding this comment.
🟡 Newly exposed constructors lack the required documentation
The two entry points foreign callers must use to create the stream managers are exported without any description (pub fn new at livekit-uniffi/src/data_stream/incoming.rs:55 and livekit-uniffi/src/data_stream/outgoing.rs:84), so the generated bindings' documentation has nothing explaining their arguments.
Impact: Users of the generated bindings get undocumented constructors, including no explanation of the optional size limit argument.
Repository rule
AGENTS.md ("API changes") requires: "New APIs should have idiomatic doc comments — All new functions and types should have at least a one-line description". Every other exported method in these two files carries a doc comment; only the #[uniffi::constructor] pub fn new items do not (livekit-uniffi/src/data_stream/incoming.rs:54-58, livekit-uniffi/src/data_stream/outgoing.rs:83-87).
| #[uniffi::constructor] | |
| pub fn new( | |
| delegate: Arc<dyn IncomingDataStreamManagerDelegate>, | |
| max_payload_byte_length: Option<u64>, | |
| ) -> Arc<Self> { | |
| /// Creates a manager, spawning its actor loop on the global runtime. | |
| /// | |
| /// `max_payload_byte_length` caps the size of any single incoming stream; the crate default is | |
| /// used when `None`. | |
| #[uniffi::constructor] | |
| pub fn new( | |
| delegate: Arc<dyn IncomingDataStreamManagerDelegate>, | |
| max_payload_byte_length: Option<u64>, | |
| ) -> Arc<Self> { |
Was this helpful? React with 👍 or 👎 to provide feedback.
| pub(crate) fn decode_data_packet(bytes: &[u8]) -> Option<ds::incoming::PacketReceived> { | ||
| let mut packet = proto::DataPacket::decode(bytes).ok()?; | ||
| let identity: common::ParticipantIdentity = packet.participant_identity.clone().into(); | ||
| let ds_packet = match packet.value.take()? { | ||
| proto::data_packet::Value::StreamHeader(header) => ds::Packet::Header { | ||
| header: header.into(), | ||
| encryption_type: common::EncryptionType::None, | ||
| }, | ||
| proto::data_packet::Value::StreamChunk(chunk) => { | ||
| ds::Packet::Chunk { chunk: chunk.into(), encryption_type: common::EncryptionType::None } | ||
| } | ||
| proto::data_packet::Value::StreamTrailer(trailer) => ds::Packet::Trailer(trailer.into()), | ||
| _ => return None, | ||
| }; | ||
| Some(ds::incoming::PacketReceived::new(ds_packet, identity)) |
There was a problem hiding this comment.
🟨 Incoming data-stream packets are treated as unencrypted regardless of the packet's encryption state
decode_data_packet hardcodes common::EncryptionType::None for every decoded header/chunk packet (livekit-uniffi/src/data_stream/common.rs:352-366), so the incoming manager's encryption-type consistency check (descriptor.encryption_type != encryption_type, livekit-data-stream/src/incoming/manager.rs:377-380) can never detect a mismatch, and encrypted payloads would be surfaced to the application as plaintext bytes. The behavior is documented as a follow-up ("the foreign side is expected to hand us already-decrypted packets"), but nothing in the FFI enforces or signals that contract.
Was this helpful? React with 👍 or 👎 to provide feedback.
Expose all the different error cases so consuming clients can get better quality errors.
This uses tokio for one (so it's not portable), but also had some security flaws (allowed paths like ../../evil.txt). So, remove it.
b0113f2 to
76f1e16
Compare
… `livekit-data-stream` (#1304) While working on #1286, I realized I made a bit of a mess of the "internal" data stream concept when extracting `livekit-data-stream` out into its own crate. This pull request attempts to address this. **Currently, what happens today:** `livekit-data-stream`'s `IncomingDataStreamManager` takes a list of `INTERNAL_DATA_STREAM_TOPICS`. These are topics used internally for v2 rpc requests / responses and messages associated with these data streams are not exposed to the user. They are filtered in two places: - Within `IncomingDataStreamManager` to conditionally expose both `InputEvent::ChunkReceived` and `InputEvent::TrailerReceived` for non internal topics - Within `RoomSession` to expose which `InputEvent::StreamOpened` events are exposed to the user. Doing this here meant that "internal" data streams could be intercepted at this level and fed into rpc / etc. This is messy - it's a little hard to follow and the filtering is split into these two places / some code duplication. What really made this untenable though was that exposing this internal state over uniffi in #1286 became particularly challenging, since you _do_ want to expose "internal" events downstream in this case - if you didn't, the uniffi consuming client couldn't handle rpc v2 / etc! **What this pull request does:** Now, `livekit-data-stream` knows nothing about "internal" or "not internal" data streams, and includes the raw `topic` of each stream in the associated event. This allows all the filtering to be centralized downstream in `RoomSession`. This fixes all of these previously mentioned problems and is a lot easier to reason about!
… FFI Add an AbortAllStreams input event to the incoming data-stream backend and expose abort_all_streams() / abort_streams_from(identity) on the UniFFI IncomingDataStreamManager, so the host can fail open readers on disconnect or participant-leave instead of letting them hang (which would also stall an ordered topic's queue). Also fix the constructor to take Option<u64> (UniFFI can't lift usize) and drop the stale reserved_topics argument to Manager::new.
4e938c0 to
38503a8
Compare
Add ByteStreamWriter/TextStreamWriter.is_open() to the UniFFI outgoing writers, backed by RawStream.is_closed(). RawStream now also marks itself closed when a chunk send fails (not only on an explicit close), so a writer whose room disconnected mid-send reports closed instead of open.
Generating Kotlin bindings for the data stream FFI produces a file that does not
compile. Nobody had done it before -- packages/kotlin has never existed in this
tree -- so this went unnoticed while Swift, Python and Node worked fine.
Two collisions, fixed differently:
- An exported Rust method named `close` collides with the non-suspend `close()`
uniffi synthesizes for AutoCloseable. They differ only by `suspend`, which
Kotlin rejects as conflicting overloads. Fixed with
[bindings.kotlin.rename], so Kotlin sees `closeStream()` and every other
language keeps `close()`.
- An error variant field named `message` collides with the `message` override
uniffi emits from Throwable, and their types differ (String vs String?) so
they cannot be merged. This one cannot be fixed from uniffi.toml: uniffi keys
the rename table by crate name but looks up enum and record members by the
item's full module path, so a rename for anything in a submodule is accepted
and silently dropped. (Methods are unaffected because they key off the crate
name -- which is why the fix above works.) There is no field-level
#[uniffi(name)] attribute either, so the field is renamed to `reason` in the
Rust source.
Renaming the field changes the exposed API, so consumers reading it by name need
updating. Swift is not one of them: its mapping binds these positionally.
Verified by generating Kotlin bindings and compiling them into the Android AAR
with no post-processing, then running that AAR's data stream tests on device.
e121d38 to
1fe0eba
Compare
Waiting for some fixes to be unstreamed before this can be swapped back over to the mainline build
This "shim" adapts the data streams v2 interface given dart's c ffi limitations (a Pointer.fromFunction callback is only invocable while the calling thread is inside a native call that Dart itself initiated).
See comments in diff explaining why, it seems to be broken in our current uniffi release version.
| @@ -6,6 +6,48 @@ android = true | |||
| package_name = "io.livekit.uniffi" | |||
There was a problem hiding this comment.
Should this be com.livekit.uniffi in the future?
There was a problem hiding this comment.
I integrated this on the Swift side (livekit/client-sdk-swift#1075) and it works end-to-end. Compression and single-packet inlining both kick in correctly once capabilities round-trip, which I verified on the wire.
Below are five gaps in the FFI surface I hit while wiring it up, ordered by what they cost a host rather than by size. None of them blocks the integration; #1 and #2 are the two that cost correctness rather than convenience, and both look cheaper to close now than after this ships and hosts have worked around them.
| #[uniffi::export(with_foreign)] | ||
| pub trait OutgoingDataStreamManagerDelegate: Send + Sync { | ||
| /// Encoded [`livekit_protocol::DataPacket`]s to be sent over the data channel transport. | ||
| fn on_packets_available(&self, packets: Vec<Bytes>); |
There was a problem hiding this comment.
The () return. You flag this at line 93 already; noting the cost so it can be weighed rather than discovered. write() can't throw when a packet doesn't reach the wire (the Swift API documents that it does), is_open() can't go false on a send failure, and there's no back-pressure — the responder acks once the host has buffered, not sent, so a producer looping on write() queues unboundedly. Host-side bounding isn't an option, since dropping stream packets is worse than the memory. A Result-returning delegate covers all three.
The Vec. It's always exactly one packet — the pump does on_packets_available(vec![packet]) per recv(). The plural signature implies batching that never happens, so every host writes a loop that runs once. Either make it singular or actually batch; batching would also make in-batch ordering free and cut the per-packet foreign round trip.
There was a problem hiding this comment.
Re batching - let me see what I can do to send multiple packets at once here. For context, my aim was to match the data track PacketsAvailable interface here which takes a Vec<Bytes>.
There was a problem hiding this comment.
For the Result so that you can pass back up send failures - I've added a Result<(), SendError> return to on_packets_available in 1dd54d0. I opted to reuse the existing SendError error type, although that has the downside of dropping any sort of "reason" type string. Maybe I could add a log or add that reason string? I'm open to others perspectives here.
For batching, as mentioned above, I've kept Vec<Bytes>, but I've changed this so that the vec sending path is used in send_text / send_bytes. With time this could be expanded but I think this is a good initial use case which also keeps it aligned with PacketsAvailable in the data track subsystem.
There was a problem hiding this comment.
Prototyped rather than just suggested — pushed as 7a2505f.
A sync callback can't deliver what the doc promises. "Return once handed to the transport" needs a synchronous transport; Swift's is async throughout, so honoring it means blocking a Rust runtime thread via a primitive that SDK forbids. The only implementable choice was to ack on buffering — so back-pressure and transport errors went unimplemented and is_open could never go false.
Four lines: #[async_trait::async_trait], async fn, .await at the call site. uniffi keeps the trait dyn-compatible itself; async-trait was already in the lockfile. Hosts get simpler too — ordering becomes structural, so Swift deleted its pump.
Verified: 104 + 12 Rust tests; 144 Swift tests clean under TSan and LIBDISPATCH_COOPERATIVE_POOL_STRICT=1; and a case untestable before — after disconnect write() throws and isOpen is false, where both previously reported success on an unwritable stream.
Unverified: Kotlin/Dart impact, async-trait now a direct dep.
… FFI The incoming manager previously signaled wire-level stream closure only via the deprecated TrailerReceived output, which the FFI layer intentionally does not forward — leaving hosts that deliver streams on ordered topics (e.g. transcription) no way to know when a stream's handler chain can advance, so a still-open stream would head-of-line-block every later stream from that sender. A trailer alone is also insufficient: inline single-packet streams never receive one. Add a StreamClosed output event emitted exactly once per opened stream on every terminal path (trailer close, inline completion, error, abort) and surface it through IncomingDataStreamManagerDelegate::on_stream_closed plus a next_closed_stream() pull on the polled adapter.
…tch one-shot sends OutgoingDataStreamManagerDelegate::on_packets_available returned (), so a packet that never reached the wire could not fail the originating call: write() couldn't throw, is_open() couldn't go false on a send failure, and the responder acked once the host had merely buffered — letting a producer looping on write() queue unboundedly. The delegate now returns Result<(), PacketDeliveryError> (a dedicated error carrying a host-provided reason, convertible into DataStreamError::Internal); throwing it fails the originating send_*/write call with SendFailed and closes the affected stream, and returning only after handing packets to the transport is what provides back-pressure. The Vec<Bytes> signature also always carried exactly one packet. Keep the shape (matching the data-track PacketsAvailable interface) but make it true: the packet channel now carries ordered batches acknowledged as a whole. One-shot sends (send_text/send_bytes) emit their entire stream — header, chunks, trailer — as a single request, i.e. one FFI crossing per send; every other call site (incremental writers, send_file's unbuffered streaming, inline sends) sends vec![packet].
…ry it in mismatch errors The FFI decode path hard-coded EncryptionType::None, and it cannot do better from the bytes alone: encrypted_packet is a member of the DataPacket.value oneof, so a host decrypting E2EE traffic replaces it with the decrypted stream packet — by the time the bytes reach the FFI, the field is absent from the wire format. That made the encryption guard in handle_chunk dead code over the FFI while still reading as active. handle_packet_received now requires the encryption type the host received (or decrypted) the packet with, making the guard live. EncryptionTypeMismatch also gains expected/received fields so hosts can report which types disagreed instead of fabricating them (Swift's error carries both).
The cap has no setter, and on the host side it typically comes from per-connection options that aren't final until connect and can differ between sessions of the same host object. The obvious host implementation — construct lazily, memoize for the object's lifetime — silently pins the cap to the first session's value. Document the intended pattern (rebuild the manager per session) instead of adding reconfiguration complexity.
There was no way to observe how many incoming streams are open, so tests exercising the abort paths had to infer "open" by signalling from inside a handler — which measures handler-dispatched rather than descriptor-registered and breaks if the two ever move relative to each other. The count is answered through the manager's input queue, so it is processed in order with previously enqueued packets: feed a header (or an abort), then await open_stream_count() to know it has landed, no sleeps or handler side-channels needed. Inline single-packet streams complete during header handling and are never counted.
…contract The delegate documents that it returns only once the packets have reached the transport, which is what orders packets, bounds a producer, and lets a failed send fail the originating `send_*`/`write`. A synchronous callback can't deliver that on hosts whose transport is async: the Swift SDK has no synchronous send path, so honoring it would mean blocking the calling thread on an async result — a Rust runtime thread, via a synchronisation primitive that SDK forbids. The practical outcome was that hosts acknowledged on buffering, leaving back-pressure and transport errors unimplemented. uniffi supports async methods on foreign traits: it applies `#[async_trait]` to the generated impl and dispatches through `foreign_async_call`, so the trait stays dyn-compatible. Awaiting the call in the pump keeps packets strictly ordered, since the next one isn't pulled until this returns. `async-trait` was already in the lockfile through livekit-api and livekit-net. The in-tree implementors — the Dart polling adapter and the test doubles — become `async fn` and are otherwise unchanged. Verified against the Swift SDK (client-sdk-swift#1075): generates `func onPacketsAvailable(packets:) async throws`, lets the host delete its ordering pump entirely, and a stream whose transport goes away now fails its `write` and reports `isOpen == false`, neither of which was observable before. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
|
Pushed 7a2505f, which makes |
handle_chunk rejects a chunk whose encryption doesn't match its stream's header, but trailers carried no encryption type at all — so in an E2EE room, an unencrypted peer could close another participant's encrypted stream cleanly and inject trailer attributes while doing so. The SDKs' hand-rolled v1 implementations (web, and the Android port's SDK-side check) validated trailers; the core did not, which blocked deleting those host-side checks. Packet::Trailer now carries the encryption type like Header and Chunk, the livekit crate threads it through Session/Engine events (it was already in hand at the emit site), and the manager rejects a mismatched trailer with EncryptionTypeMismatch before merging its attributes.
The arm64-v8a slice is 1490 KiB with the data streams v2 surface (async foreign-trait dispatch, stream-closed and open-stream-count, trailer encryption checks), over the 1280 KiB budget set in May before that work. cargo's strip = "symbols" is already applied to the artifact, so AGP's strip pass has nothing further to remove — the growth is real surface, landed intentionally.
|
FYI @pblazej - 1da01b4 was an issue I found when updating the android implementation to take into account the changes from your review. I think it may have some implications on the swift version as it changes the format of the |
3 participants
Adds data streams v2 to the
livekit-unifficrate, and exposes it fairly similarly to how data tracks works.I've also for now added a python test script (It's the most complete bindgen set up in this project which I am familiar with) which exercises this api as an example of what it would look like in practice. I'll remove this before an eventual merge, but I thought it would be useful for reviewers:
$ python3 datastream_uniffi_test.py --- OUTGOING: PACKETS: [b'jP\n$4a501804-960c-4d54-94df-81784ffab6b1\x10\xc6\xa5\xb0\xaf\xf93\x1a\x04test"\ntext/plain(\x0bJ\x00Z\x0bhello world'] --- INCOMING: TEXT STREAM OPENED: alice CONTENTS: hello world