Repository navigation
Data streams v2 - #1166
Draft
1egoman wants to merge 29 commits into
Draft
Data streams v2#11661egoman wants to merge 29 commits into
1egoman wants to merge 29 commits into
Conversation
Groundwork for delivering the Rust core through Dart's Native Assets, which is only stable from Flutter 3.38 / Dart 3.10 onwards. The newer language version turns on lints the tree does not yet satisfy: - `*.g.dart` joins the analyzer exclusions. AGENTS.md already described json_serializable output as excluded, but only the protobuf patterns actually were; 47 of the 48 new `use_null_aware_elements` hits are in generated files nobody edits by hand. - The remaining `use_null_aware_elements` hit, in `Room`, and four `unnecessary_underscores` hits in the RPC tests, are fixed directly. Crossing Dart 3.7 also switches `dart format` to the tall style, which reformats the whole tree. That is mechanical and lands separately, in the commit that follows, so it does not bury this one. `dart format --set-exit-if-changed` is therefore red at this commit and green again at the next. No functional change.
Pure output of `dart format .`. No functional or semantic change. The formatter picks its style from the package language version, so the SDK bump in the previous commit crosses Dart 3.7 and switches the whole tree from the short style to the tall one. `dart format --set-exit-if-changed` is CI-enforced, so this has to land -- it is split out only so the bump and its lint fixes stay readable. Reproduce with `dart format .` on the parent commit.
Establishes the development loop for calling into `livekit-uniffi` from this SDK, and exercises it with `buildVersion()` -- a synchronous, argument-free call returning a string, so a green test means the whole chain is intact: the build hook resolved a cdylib for the target, Native Assets bundled it, `@Native` bound the symbol, and a value crossed back. `lib/src/uniffi/` splits native/web with the same conditional-import pattern as the rest of the SDK. There is no dynamic library to load on the web, so `uniffi_io.dart` is the only file permitted to import the generated bindings -- reaching them from anywhere web-reachable pulls `dart:ffi` into a web compile and breaks `flutter build web`/`--wasm`. Callers branch on `LiveKitUniffi.isAvailable`. The facade is internal for now; nothing is added to the public API surface. `livekit_uniffi` is not on pub.dev, so both pubspecs override it to a sibling rust-sdks checkout -- overrides do not propagate from a dependency, hence both. Until it is published this branch cannot merge: CI has no rust-sdks checkout, so `flutter pub get` will fail there. Verified: 392 tests pass, web and wasm builds succeed, analyze and format are clean.
Swaps the native data-stream internals onto `livekit-uniffi`, which implements data streams v2: single-packet inline sends for small finite payloads, deflate-raw compression, UTF-8-aware chunking and MTU-bounded headers. Reimplementing that in Dart would have meant a third copy of a wire protocol that has to stay byte-compatible with the JS, Swift and Android SDKs; this way the framing has exactly one implementation. Mirrors the Swift SDK's `data-streams-v2` branch: one `DataStreams` subsystem per Room, created eagerly and outliving connect/disconnect so handler registrations survive a reconnect, with the FFI boundary being serialized `DataPacket` bytes in both directions. Outbound packets go back through `Engine.sendDataPacket`, so E2EE wrapping, reliable sequencing and resume-resend are all unchanged. Web keeps the existing Dart implementation, moved verbatim behind the same interface: the core ships a cdylib and cannot run in a browser, and RPC v2 rides on text streams. Web advertises `clientProtocol` v1 and no capabilities, so v2 senders fall back to uncompressed multi-packet for it — interop is safe by spec. Two findings worth recording, both now in AGENTS.md: - The core's *push* delegates cannot be used from Dart at all. uniffi compiles a callback interface to `Pointer.fromFunction`, valid only on the isolate's thread, and the core invokes those from its tokio runtime — the VM aborts with "Cannot invoke native callback outside an isolate". Both managers are constructed with `delegate: null` and drained via the new `nextPackets`/`nextOpenedStream` pull API instead. `RemoteParticipantRegistryDelegate` is the one safe callback: it is only called inside a `send*` future, which uniffi polls from the calling thread. - Only the pump may dispose a uniffi reader. Disposing from a subscription's `onCancel` frees the Rust handle while a `next()` is in flight, which surfaces as a SIGBUS with no Dart stack. Public API is unchanged; `sendBytes`, a `compress` option, `ClientCapability`, `Participant.capabilities` and `ClientProtocolVersion.v2` are additive. Protobufs are regenerated — `DataStream.Header` previously had no `inlineContent`/`compression` and `ClientInfo.Capability` stopped at `CAP_PACKET_TRAILER`. Behavior changes, all consequences of the core owning the framing: `onProgress` loses per-chunk granularity, transport send errors no longer reach `write()`, `encryptionTypeMismatch` is unreachable, `ByteStreamReader.readAll()` no longer silently de-duplicates identical chunks (it collected into a Set), and `sendText`'s `totalLength` is now the UTF-8 length rather than `codeUnits.length`.
Replaces the bespoke pull methods added to `OutgoingDataStreamManager`
and `IncomingDataStreamManager` with a composed adapter layer, so the
data-stream code itself goes back to exactly what it was.
The managers push their output to a foreign delegate from a tokio
thread, which Dart cannot accept. Rather than teach each manager a
second, pull-shaped mode, `data_stream/polled.rs` implements those same
delegate traits *in Rust*, buffers into a channel, and exposes an
`async fn next_*` the foreign side awaits. The delegate still runs on a
tokio thread — which is fine precisely because it is Rust, so nothing
crosses the FFI until the await resolves.
Three things this buys over the previous shape:
- `incoming.rs` and `outgoing.rs` are untouched, so Swift and Kotlin are
provably unaffected — not even an `Option` added to a constructor.
- The two halves are wired together inside Rust by
`polled_{outgoing,incoming}_data_stream_manager`, so no binding has to
pass a Rust object where `Arc<dyn Trait>` is expected — which
uniffi-dart cannot currently do anyway.
- The pattern generalizes. Data tracks has the identical
delegate-on-a-tokio-thread problem and can reuse it.
Not feature-gated for now: the module is small and inert, and gating it
would fork the cdylib build matrix in uniffi-cdylib.yml.
Also closes a lifetime hazard the previous shape shared: the queues now
expose `close()`, which wakes a pending `next_*` with `None` so its pump
can exit and dispose the queue itself. Disposing from `dispose()` while
a pump was blocked would have freed the handle mid-call — the same
use-after-free that showed up as a SIGBUS in the reader path.
`connect` opened with `var roomOptions = this.roomOptions;` while a parameter of the same name was in scope. Dart permits a local to shadow a parameter — silently, with no warning and the local winning — so every `RoomOptions` a caller passed to `connect` was discarded and the Room's own options were used instead. Nothing surfaced the mismatch. The parameter is deprecated in favour of the `Room` constructor, but deprecated is not the same as inert: while it is still accepted it has to take effect. It now does, falling back to the Room's options when absent. `Engine.connect` already adopted whatever it was handed, so the value propagates from there without further changes. The local is renamed to `effectiveRoomOptions`, since restoring the parameter's visibility is the whole point and leaving two things called `roomOptions` in one scope is what caused this. Adds a regression test, and threads `connectOptions`/`roomOptions` through the E2E container so it can be exercised. Verified the test fails against the old shadowing behavior.
The Rust core has always accepted a payload cap — it is the guard against a sender making a receiver allocate arbitrary memory, including via a compressed stream that inflates far past its wire size — but nothing exposed it, so `null` was hardcoded and every room ran with the core's 5 GB default. Placed on `ConnectOptions` rather than `RoomOptions`, which is where Swift puts it. It fits the lifecycle better here: the incoming manager is created lazily on the first inbound packet precisely so a connect-time value is in effect by the time it is read. Worth knowing it diverges from Swift if cross-SDK consistency matters more than that. Also enforced on web, which otherwise would have accepted the option and silently done nothing with it — the same trap as the ignored `connect(roomOptions:)` argument. The Dart path tracks accumulated content bytes per stream so an unknown-length stream is capped too, not just one that declares an oversized `totalLength`. Both paths follow the core's semantics, which are subtler than they look: the stream-opened event fires *before* the cap is applied, so the topic handler still runs and it is the reader that fails with `LengthExceeded`. A consumer is told the stream died rather than watching it never arrive. My first pass had web refusing the stream outright and the doc comment describing that; both are corrected here.
1egoman
force-pushed
the
data-streams-v2
branch
from
August 10, 2026 20:32
701f538 to
60b93c2
Compare
hiroshihorie
pushed a commit
that referenced
this pull request
Aug 11, 2026
… variable, NOT implicitly this.roomOptions (#1169) While working on #1166, I discovered this bug incidentally, so here is an isolated fix. `connect`'s first line is `var roomOptions = this.roomOptions;`. However, a parameter of the same name (`roomOptions`) was in scope. After some unexpected behavior, I realized that dart permits a local to shadow a parameter (silently, with no warning and the local winning) so every `RoomOptions` a caller passed to `connect` was discarded and the Room's own options were used instead. Nothing surfaced the mismatch. The `roomOptions` parameter on `connect` seems to be deprecated in favor of the parameter on the `Room` constructor, but deprecated is not the same as inert: while it is still accepted it has to take effect. It now does, falling back to the Room's options when absent. `Engine.connect` already adopted whatever it was handed, so the value propagates from there without further changes. The local is renamed to `effectiveRoomOptions`, since restoring the parameter's visibility is the whole point and leaving two things called `roomOptions` in one scope is what caused this. I also addeds a regression test, and threaded `connectOptions`/`roomOptions` through the E2E container so it can be exercised. I verified the test failed against the old shadowing behavior. Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
# Conflicts: # example/pubspec.yaml # lib/src/core/engine.dart # lib/src/core/room.dart # lib/src/e2ee/key_provider.dart # lib/src/extensions.dart # lib/src/json/agent_attributes.g.dart # lib/src/token_source/caching.g.dart # lib/src/token_source/jwt.g.dart # lib/src/token_source/room_configuration.g.dart # lib/src/token_source/token_source.g.dart # pubspec.lock # pubspec.yaml
… core handleIncomingPacket received the encryption type from the engine and dropped it. The core now requires it — it cannot be recovered from the bytes, since decryption replaces the encrypted_packet oneof member — and uses it to hold every stream to the encryption its header arrived under, failing mismatched chunks and trailers with EncryptionTypeMismatch (whose expected/received detail is surfaced in the error message). Incoming stream infos are now stamped with the encryption the stream actually arrived under, as reported by the core, instead of the room's own setting — a plaintext stream is reported as plaintext even in an encrypted room. Outgoing infos keep the room-derived value: payload crypto happens in the engine after the core.
The incoming pull adapter grew a second channel: nextClosedStream() yields a notification once per opened stream (trailer, inline completion, failure, or abort). Nothing consumes the signal yet — it exists for ordered per-topic delivery, which this SDK doesn't implement — but the channel is unbounded on the Rust side, so leaving it undrained would grow memory with every stream received for the manager's lifetime. A discard pump drains it. The two pumps share one queue object, so disposal moves out of the incoming pump: the queue is disposed exactly once, after both pumps have exited (close() wakes them both). Disposing from either pump's own exit path would free the handle while the other might still have a call in flight.
The incoming manager's payload cap is fixed at its construction, and the ConnectOptions.dataStream it comes from can change between connects of the same Room. reset() only aborted open streams and reused the manager, silently pinning the first session's cap — exactly the hazard the core's constructor docs call out. reset() now discards the manager: abort open streams so blocked readers error, close the queue so both pumps wake and dispose it, and release the manager. The next inbound packet lazily builds a fresh one against the current connect options; handler registrations live on this class and survive.
- The uniffi smoke suite now drives the polled incoming manager end to end: 2-arg handlePacketReceived, the in-order openStreamCount query, and the stream-closed queue. It is the gate that fails first when the vendored bindings fall behind the Rust crate. - v2 tests: a chunk that changes encryption fails its reader with EncryptionTypeMismatch (with the expected/received detail), stream info reports the arrival encryption rather than the room's setting, and open stream accounting is observable (including that reset() discards the manager). - debugOpenStreamCount on NativeDataStreams: answered on the core's loop in order with previously fed packets, so tests wait deterministically instead of polling.
livekit_uniffi 0.1.10 is published, and its release carries the prebuilt libraries the build hook downloads, so the path overrides to a sibling rust-sdks checkout are no longer needed in either pubspec.
The path override becomes the opt-in for developing against unreleased crate changes rather than the default, and the note about releases lacking assets is replaced with how download mode behaves.
flutter_webrtc moved to 1.6.2+hotfix.1 on main.
0.1.12 is the first version published by the release pipeline. Its build hook keeps each target architecture's downloaded library in its own directory, which the universal macOS release build needs; 0.1.10 wrote both to one path and the macOS CI job failed in lipo. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
# Conflicts: # pubspec.lock
The example logs the Rust core build version at startup and shows it on the connect page, one synchronous FFI call that proves the native library was bundled and loads on the platform.
The change adds a public API and a dependency whose build hook downloads native libraries during consumer builds, which is a feature release rather than a patch. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
The automatic merge of pubspec.lock dropped the livekit_uniffi, archive and posix entries that only this branch resolves. Regenerated with flutter pub get. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
changt
reviewed
Oct 16, 2026
| reliability: Reliability.reliable, | ||
| ); | ||
| } catch (e) { | ||
| // The core acknowledges sends unconditionally, so there is nobody to propagate this to. |
Contributor
There was a problem hiding this comment.
This silently swallows send failures. I believe the prior version exposed failures to the caller.
| /// The pump owns the queue's lifetime and is the only thing that may dispose it — freeing it | ||
| /// while a `nextPackets` is in flight is a use-after-free. [dispose] wakes us by closing the | ||
| /// queue rather than releasing it. | ||
| Future<void> _pumpOutgoing() async { |
Contributor
There was a problem hiding this comment.
should this waitForBufferStatusLow?
# Conflicts: # AGENTS.md # test/uniffi/uniffi_test.dart
|
Caution Breaking change detected without major changeset
If this is intentional, please add a changeset with |
This branch has not been deployed
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Initial port of data streams v2 to the flutter sdk. This is built on top of #1160, and that PR must be merged first.
Summary
Background context on data streams v2
Previously, the flutter sdk had its own data streams implementation. 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.
In the corresponding swift and android pull requests, I was able to fully cut over to data streams v2. However, the flutter sdk also has to support web, and today uniffi-dart does not support WASM - see here for an early issue on the bindgen side tracking this work.
So in lieu of fully cutting over:
Note that it could be possible to make flutter web use the web sdk's data stream v2 implementation, but it looks like it would introduce a lot of infrastructure complexity. Given this and that it seems like flutter web support is somewhat experimental anyway, I'm tempted to keep it on the existing implementation.
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 new data stream errors types
The old flutter specific data streams implementation wasn't quite as strict and didn't surface as many error cases to the caller which were encountered while reading the stream as the rust implementation now does. This needs some testing in some of these edge cases to make sure I didn't inadvertently make this backwards incompatible in a non-aceptable way.
Warning
This pull request was LLM generated and has only been reviewed by a human who isn't a domain expert in android development. I have tested this and confirms it works in the happy path, but no other validation has been done.
A more thorough review of this needs to occur before it could be merged.
Todo