Data streams v2 - #997
Conversation
Data streams v2 needs three things that landed upstream in protocol #1621
(first tagged v1.46.8); this SDK was pinned at v1.45.8-29:
- ClientInfo.Capability.CAP_COMPRESSION_DEFLATE_RAW = 2 (advertise)
- ParticipantInfo.capabilities = 21 (read remote caps)
- DataStream.Header.inline_content / compression (v2 wire fields)
Pinned to 28e604c, the same commit client-sdk-swift@data-streams-v2 and
rust-sdks use, so all three SDKs agree on the wire contract.
Two unrelated breaks from crossing five minor protocol versions, both fixed
here so the bump lands green:
- SignalClient's messageCase `when` gained STORE_DATA_BLOB_RESPONSE and
GET_DATA_BLOB_RESPONSE; stubbed as TODO like their neighbours.
- RoomAgentDispatch gained an `attributes` map, which ProtoConverterTest
requires the Kotlin DTO to mirror. Added as a nullable field with a
default, so it is source- and serialization-compatible.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Adds the livekit-uniffi dependency to the SDK and everything needed to exercise
it from a host JVM test run. No SDK behavior changes yet.
The artifact is unreleased, so it resolves from mavenLocal() (already enabled on
this branch). Build it with `cargo make android-package-local` in rust-sdks, or
for a JVM-test-only workflow just generate the Kotlin bindings plus a host
cdylib -- no NDK required, and it avoids three Android target dirs' worth of
disk.
Three obstacles found and handled along the way:
- Nobody had ever generated Kotlin bindings for the data stream FFI
(`packages/kotlin/` did not exist in rust-sdks), and doing so surfaces two
uniffi codegen bugs that stop the generated file compiling at all:
a Rust method named `close` collides with the AutoCloseable `close()` uniffi
synthesizes, and an error variant field named `message` collides with
Throwable.message. scripts/patch-uniffi-kotlin.py patches the generated
output -- a build artifact, so rust-sdks source is untouched -- following the
precedent of the existing swift-workarounds task. Both still need a real
upstream fix.
- livekit-uniffi's AAR declares minSdk 24 against this SDK's 21. The native
library is built for platform 21, so the declaration is the only conflict;
overridden via tools:overrideLibrary in both manifests.
- The AAR depends on jna's *aar*, which carries only Android dispatch
libraries, so host JVM tests died in JNA before ever reaching our code. The
plain jna jar is added as a test dependency for its desktop libjnidispatch.
gradle/uniffi-native-lib.gradle points JNA at a host build of the library,
auto-discovering a sibling rust-sdks checkout and overridable via
LIVEKIT_UNIFFI_LIB_DIR. UniffiNativeLibraryTest asserts the library loads and
its checksums match the bindings, so a broken setup fails once with a clear
message instead of once per data stream test.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Adds the types and read path for per-peer feature negotiation, which the data
streams v2 send path needs in order to decide whether a given recipient can
accept an inline or compressed stream.
- ClientCapability: public enum mirroring ClientInfo.Capability. fromProto
returns null for unrecognized values instead of throwing, unlike most
fromProto helpers here -- capabilities are an open set, so a peer on a newer
SDK must be tolerated, not fatal.
- ClientProtocolVersion.DATA_STREAM_V2 (2), documented as a baseline
commitment rather than an optional feature.
- Participant.capabilities, populated from ParticipantInfo.capabilities
alongside the existing clientProtocol.
Purely additive, and deliberately read-only for now: this SDK does not yet
advertise v2 or any capability. Advertising has to wait until the receive path
can actually handle inline and compressed streams, otherwise peers would
start sending framings the current Kotlin implementation cannot parse. That
flip lands with the cutover.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Introduces the single place that touches livekit-uniffi. Not wired into anything
yet -- nothing injects it, so this commit changes no behavior.
DataStreams owns both FFI managers, the topic to handler registry, the FFI
delegates and the error mapping, so that everything above it keeps dealing only
in this SDK's own types.
Notable choices:
- The outgoing manager is built eagerly; the incoming one lazily on the first
inbound packet. Its payload cap comes from RoomOptions, which is not final
until connect() -- after this class is constructed -- so reading it eagerly
would silently ignore a maxPayloadSize passed to connect().
- Outbound packets go through an unbounded channel drained by one coroutine.
The FFI delegate is a synchronous callback on a Rust runtime thread and can
neither block nor suspend, but sending has to await publisher connection and
data channel backpressure. This keeps emission order and restores the
backpressure the previous implementation had.
- Stream handlers are dispatched onto our own scope rather than run inline on
the FFI thread, so an app handler that blocks cannot stall the core's runtime
and with it every other incoming stream.
- Delegates hold their owner strongly, unlike Swift. The JVM collects cycles,
so Swift's weak back-reference buys nothing here. What does matter is close()
running: the FFI's handle map holds the delegates from a static root, so
DataStreams registers with CloseableManager to release the native handles.
- Room state the send path needs (remote identities, protocols, capabilities,
the payload cap) arrives as assignable lambdas rather than by injecting Room,
which would be a Dagger cycle. Same pattern Room already uses for the RPC
managers.
Also adds the additive options this needs: `compress` on both stream option
classes and RoomOptions.dataStreamOptions.maxPayloadSize, all defaulted to
previous behavior.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Cuts the data stream implementation over to livekit-uniffi. The hand-written
packet building and stream reassembly are gone; the managers are now thin
adapters over DataStreams.
The public API is unchanged. IncomingDataStreamManager and
OutgoingDataStreamManager keep their interfaces, so Room's and LocalParticipant's
delegation is untouched, and TextStreamSender / ByteStreamSender /
TextStreamReceiver / ByteStreamReceiver keep working by reusing their existing
seams: an FFI-backed StreamDestination, and a Channel pumped from an FFI reader.
The interface's handleStreamHeader/handleDataChunk/handleStreamTrailer remain as
shims that rebuild a packet, though Room now forwards whole packets instead --
v2 headers carry inline content and compression that only the core reads.
Behavior changes fall out of the implementation moving into a Rust actor, and
existing tests were updated to match rather than papered over:
- Sending and receiving are now asynchronous. A completed write means the core
accepted the payload, not that it reached the wire, and an incoming stream is
delivered after a round trip through the core. Tests that asserted
synchronously now await the outcome.
- Send failures no longer reach the caller: the core acknowledges a send when
it hands the packets over.
- StreamException.EncryptionTypeMismatch is unreachable; the core normalizes
encryption type at the boundary.
Three problems this shook out, all now handled:
- Everything on the FFI boundary runs on a real dispatcher, never the caller's.
The core resumes calls from its own runtime threads, which a virtual-time
test dispatcher can never deliver; worse, an unconfined dispatcher resumed
our coroutines *inline on a core runtime thread*, where waiting on data
channel backpressure deadlocked the runtime and stopped every stream.
- MockDataChannel's buffers became copy-on-write; sends now genuinely arrive
from several threads and assertions iterate the list concurrently.
- Test classes each get their own JVM. The core is process-global state loaded
via JNA, and Robolectric's per-class classloaders re-initializing the
bindings left callbacks pointing nowhere -- whole classes would time out
depending on execution order. Costs ~90s on this module; see the comment in
livekit-android-test/build.gradle.
All 326 tests pass, verified stable over repeated cold runs.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
…ith tests
Completes the cutover by telling peers what we can now do, and adds the tests
for the parts of it that only fail in interop.
Advertisement, deferred until the receive path could handle what it invites:
- ConnectOptions.clientProtocol now defaults to DATA_STREAM_V2.
- The connect URL carries a `capabilities` param, and ClientInfo the matching
repeated field, both sourced from one ADVERTISED_CLIENT_CAPABILITIES list.
Compression is advertised unconditionally, since it is done by the core
rather than a platform codec.
Tests (54 new, 380 total, all passing):
- DataStreamsV2SendTest walks the framing matrix from the spec's "Minimum
required test cases" -- pre-v2 room, all-v2 room, v2-without-the-capability,
mixed room, targeted subsets, compress opt-out, incremental writers -- and
asserts on the packets that reach the engine. These are really tests of our
registry wiring: the core picks the framing from what we report about each
recipient, so getting that wrong produces packets a peer cannot read while
everything still looks fine locally.
- DataStreamsV2ReceiveTest covers the framings only v2 produces (inline,
inline compressed, a deflate stream spread across chunks), plus topic
routing, sender identity, abort-on-disconnect, the payload cap, and a
multi-byte text round trip through the byte channel the public reader uses.
- DataStreamsConversionTest pins the translation layer, in particular the
error mapping, which is lossy by design -- several core failures fold onto
one pre-existing public exception, and a wrong fold is invisible.
- ConnectionParamsTest asserts the advertisement on the actual connect URL.
The Swift SDK shipped this wiring broken on one of its two connect paths, so
it is asserted on the wire rather than on the values feeding it.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
spotlessCheck is a CI gate; this is its output plus one flake fix. awaitJob waited only for the RPC itself, but a completed call can still have siblings finishing behind it -- closing the request stream, emitting a disconnect event -- whose continuations are posted back to the test dispatcher from the core's threads. Occasionally one landed after the test body returned and surfaced as "unfinished coroutines found during the tear-down". Pump briefly after the job completes so they run inside the test. Verified over four consecutive cold runs of the full suite. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
|
Every failure the core can report is now distinguishable, rather than several
folding onto the nearest pre-existing case.
New exceptions:
- HeaderTooLargeException, PayloadTooLargeException. Both are subclasses of
LengthExceededException rather than siblings, so existing code catching that
keeps catching every size-limit failure while new code can tell them apart.
Neither failure mode existed before v2 (there was no header budget and no
payload cap), so nothing was relying on the old folding.
- InternalException, which previously arrived as a TerminatedException.
TerminatedException gains a `reason`, defaulted and @jvmoverloads'd so
single-argument construction is unchanged. It separates the five remaining cases
that share the type -- already closed, invalid header, missed chunk, send
failed, invalid file name -- plus IO.
Io no longer maps to AbnormalEndException. A local file read failing is not the
remote closing the stream on us, which is what that exception documents; it is
now TerminatedException with reason IO.
The one caveat: StreamException is sealed, so an exhaustive `when` over it in
consumer code will need a branch for InternalException. Catch-based handling,
which is how exceptions are used here, is unaffected.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Surfacing that field was fallout from the protocol bump, not part of data streams, and it does not belong in this change. ProtoConverterTest requires every proto field to be mirrored on the Kotlin DTO, which is why it was added. Whitelisted instead, alongside the fields already listed there, so the test states plainly that it is not surfaced yet. Worth knowing: the SDK therefore cannot set agent dispatch attributes, which the server now accepts. That is a real gap, just an unrelated one -- if it is wanted, it should be its own change. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
…blocker
Adds instrumented tests and, in running them, found that the SDK could not have
worked on a 16 KB page-size device at all.
livekit-uniffi's AAR pins jna 5.16.0, whose libjnidispatch.so fails to load
there:
E linker: ".../libjnidispatch.so" program alignment (8192) cannot be smaller
than system page size (16384)
which surfaces as an inscrutable `NoClassDefFoundError: com.sun.jna.Native` --
JNA's classpath fallback masking the real dlopen failure. Our own
liblivekit_uniffi.so is fine (NDK r27 aligns to 16 KB by default), and so is
WebRTC's; JNA's prebuilt library is the only one that fails. Verified by loading
all three directly: only jnidispatch failed, and jna 5.19.1 loads.
The SDK now overrides the transitive pin. This matters beyond the emulator:
Android requires apps targeting API 35+ to support 16 KB page sizes, and such
devices ship today, so every data stream would have died on the first FFI call.
The real fix belongs in livekit-uniffi's own build.
DataStreamsOnDeviceTest (10 tests, passing on an API 37 arm64 emulator) covers
what the host JVM tests cannot:
- the AAR's .so loading on Android, through JNA, from packaged jniLibs;
- the bindings' Android cleaner path, chosen at API 34+, which uses
android.system.SystemCleaner. Under Robolectric that throws
IllegalAccessError and needs a JVM flag, so this is the only place it runs as
written;
- the v2 framings produced on-device matching the host build's, including a
send-to-receive loopback through the core that reconstructs a compressed
inline payload;
- this SDK's conversions and error mapping in an Android runtime.
No mocking framework: androidTest has no Mockito here, so these drive the FFI
directly with a capturing delegate, the same seam client-sdk-swift's tests use.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Replaces scripts/patch-uniffi-kotlin.py, which rewrote the generated bindings
after the fact, with fixes in livekit-uniffi itself. The bindings now compile as
generated, so there is no post-processing step to remember or keep working.
The two collisions needed different mechanisms:
- `close` is fixed by [bindings.kotlin.rename] in livekit-uniffi's uniffi.toml,
which is per-language exactly as wanted: Kotlin sees `closeStream()` while
Swift, Python and Node keep `close()`. Only this SDK's two call sites change.
- `message` could not be. UniFFI keys its rename table by crate name but looks
up enum and record *members* by the item's full module path, so a rename for
anything declared in a submodule is accepted and silently ignored -- which is
also why the method rename works, since methods key off the crate name. That
looks like an upstream bug and is worth reporting. There is no field-level
`#[uniffi(name)]` attribute in 0.31 either (uniffi_macros takes field names
straight from the Rust identifier), so the field is renamed to `reason` in
Rust. That is global rather than Kotlin-only, but it is a better name for an
error detail anyway, and Swift binds these positionally so its mapping is
unaffected.
Renaming the field changes the FFI metadata, so the AAR and all three Android
libraries were rebuilt; the checksum check between bindings and .so would fail
otherwise.
Verified: 380 unit tests and 10 instrumented tests on an API 37 emulator, both
against bindings generated with no manual edits.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Every debug launch of sample-app dies in Activity.onCreate:
java.lang.AbstractMethodError: abstract method "androidx.lifecycle.ViewModel
androidx.lifecycle.ViewModelProvider$Factory.create(kotlin.reflect.KClass,
androidx.lifecycle.viewmodel.CreationExtras)"
on receiver leakcanary.internal.ViewModelClearedWatcher$...
LeakCanary watches ViewModels by registering a ViewModelProvider.Factory, and it
is binary-incompatible with the lifecycle 2.8.0 this project resolves: 2.8.0's
KMP refactor made `create(KClass, CreationExtras)` part of the interface, and
LeakCanary implements only the older overload. Checked 2.14 as well as the
pinned 2.8.1 -- both crash -- so this is not a stale-version problem and there
is no version to bump to.
Removing the auto-install provider is LeakCanary's own documented way to turn it
off, and it is the smallest change that gets the app running. It is scoped to
sample-app's debug manifest, so nothing else is affected.
Pre-existing and unrelated to data streams -- the same dependency is on main --
but it blocks running the sample app at all, which is where the data streams
panel in the next commit lives.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Adds a developer panel for exercising data streams by hand, reached from a new icon beside the RPC tester in the in-call controls. Conceptually the Android counterpart of livekit-examples/rust-dev-client#18, and structured the same way that PR describes: a send section and a subscriptions section. Until now the sample could only send a fixed `lk.chat` text stream from an AlertDialog and surfaced receipts as a Toast -- no topic, no destination, no byte streams, and nowhere to watch what arrived. Send: text or bytes, a topic, a destination (a remote participant or everyone), and a content box with `hello world` and `20k random` presets. Bytes are the UTF-8 of the same box. The result line reports the new stream's id, or the error. The 20k preset is deliberately random rather than a repeated character: random data does not compress, so it exercises the compressed multi-packet path instead of collapsing into a single inline packet. Subscribe: register a topic as text or bytes and watch it fill up. Each subscription is a card with its own scrolling list of arrivals, newest first, showing sender, size, time and a preview -- truncated for text, hex plus utf8 for bytes, and capped at 100 per topic. State lives on CallViewModel next to the RPC tester's, so subscriptions keep collecting while the panel is closed and are unregistered with the room in onCleared. Subscriptions are keyed on (topic, kind) because text and byte handlers are separate registries and the same topic can carry one of each. A topic something else already owns -- `lk.chat`, or the RPC topics the Room registers -- comes back as a failed Result and is shown, not thrown. Read failures are recorded as the preview, so a PayloadTooLarge or an aborted sender is visible rather than silent. Verified against a local livekit-server with two emulators in one room: text round-tripped, 20k random arrived as 19.5KB with the preview truncated, bytes and text coexisted on one topic with the hex/utf8 preview correct, the destination dropdown listed the peer, and subscribing to `lk.chat` was refused without a crash. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
![devin-ai-integration[bot]](https://avatars.githubusercontent.com/in/811515?s=60&v=4){kind=small}
| // Forwarded whole rather than destructured: the core re-decodes the packet itself, and v2 | ||
| // headers carry fields (inline content, compression) that only it interprets. The packet has | ||
| // already been decrypted by the engine, which is what the core expects. | ||
| dataStreams.handleIncoming(dp) |
There was a problem hiding this comment.
🔴 Unencrypted data streams are accepted and reported as encrypted when end-to-end encryption is on
The actual encryption of each received stream packet is thrown away when the packet is handed to the new stream engine (dataStreams.handleIncoming(dp) at livekit-android-sdk/src/main/java/io/livekit/android/room/Room.kt:1368), and every stream is instead labelled with whatever the local room is configured for, so a stream that arrived unencrypted is delivered to the app as if it were encrypted.
Impact: With end-to-end encryption enabled, plaintext data streams from a peer are silently accepted and shown to the app as encrypted, instead of failing the stream.
How the per-packet encryption type is lost
RTCEngine computes the real per-packet encryption at livekit-android-sdk/src/main/java/io/livekit/android/room/RTCEngine.kt:1340-1347 (NONE unless the packet actually carried an encryptedPacket) and passes it to onDataStreamPacket.
Before this PR, IncomingDataStreamManagerImpl.handleDataChunk/handleStreamTrailer compared that value against the encryption type recorded on the stream header and closed the reader with StreamException.EncryptionTypeMismatch on a mismatch.
Now Room.onDataStreamPacket ignores its encryptionType argument entirely, and DataStreams.currentEncryptionType() (livekit-android-sdk/src/main/java/io/livekit/android/room/datastream/DataStreams.kt:313-319) derives the value purely from engine.e2EEManager?.isDataChannelEncryptionEnabled() — i.e. the local configuration, not the received bytes. The core cannot supply it either; the same comment notes the core always reports NONE.
The same loss applies to the legacy shims IncomingDataStreamManagerImpl.handleStreamHeader/handleDataChunk/handleStreamTrailer, which now discard their encryptionType parameter.
Prompt for agents
Incoming data stream packets used to be validated against the encryption type reported by RTCEngine: the stream header recorded an encryption type, and every subsequent chunk/trailer whose packet-level encryption type differed closed the reader with StreamException.EncryptionTypeMismatch. After the port to the Rust core, Room.onDataStreamPacket forwards the whole DataPacket and drops the encryptionType argument, and DataStreams.currentEncryptionType() reports the local room's e2ee configuration rather than what actually arrived. The result is that an unencrypted stream received while e2ee is enabled is accepted and its StreamInfo.encryptionType is reported as GCM.
Consider re-introducing the check in the SDK layer, since the core cannot see transport encryption: track the encryption type observed on a stream's header (keyed by stream id) inside DataStreams, compare it against each subsequent packet's encryption type before feeding the packet to the core, and abort the stream (surfacing StreamException.EncryptionTypeMismatch) on a mismatch. Also use that observed value, rather than the room's current configuration, when stamping StreamInfo.encryptionType for incoming streams.
Was this helpful? React with 👍 or 👎 to provide feedback.
| * | ||
| * The FFI delegate is a plain synchronous callback on a Rust runtime thread: it can neither | ||
| * block nor suspend, but sending has to await publisher connection and data channel | ||
| * backpressure. Handing off through an unbounded channel drained by a single coroutine keeps | ||
| * packets in the order the core emitted them while restoring the backpressure the previous | ||
| * implementation had. | ||
| */ | ||
| private val outboundPackets = Channel<ByteArray>(Channel.UNLIMITED) |
There was a problem hiding this comment.
🟡 Large outgoing data streams can be buffered entirely in memory
Outgoing stream packets are placed on a queue with no size limit (Channel<ByteArray>(Channel.UNLIMITED) at livekit-android-sdk/src/main/java/io/livekit/android/room/datastream/DataStreams.kt:144) while the sender is told the write succeeded immediately, so a large send can pile up in memory faster than the network drains it.
Impact: Sending a large file or payload can grow memory without bound and risk an out-of-memory crash.
Why the previous backpressure no longer reaches the producer
Previously ManagerStreamDestination.write chunked the payload itself and, for every chunk, awaited engine.waitForBufferStatusLow(...) followed by engine.sendData(...) before returning — so a slow data channel suspended the caller.
Now WriterDestination.write (livekit-android-sdk/src/main/java/io/livekit/android/room/datastream/DataStreams.kt:504-512) returns as soon as the core accepts the write, and the core pushes encoded packets synchronously through OutgoingDelegate.onPacketsAvailable (.../DataStreams.kt:344-353) using trySend onto the unlimited channel. Only the single draining coroutine at .../DataStreams.kt:286-301 waits on waitForBufferStatusLow; nothing throttles the producer.
This is most visible for sendFile, where the core reads the whole file rather than the caller streaming it chunk by chunk. The comment at .../DataStreams.kt:137-143 claims this "restor[es] the backpressure the previous implementation had", which an unbounded channel does not do — it only preserves ordering.
Prompt for agents
DataStreams hands outgoing packets from the FFI delegate to a Channel with UNLIMITED capacity, and the only place that waits on engine.waitForBufferStatusLow is the single consumer coroutine. Because the FFI delegate callback is synchronous and cannot block or suspend, trySend always succeeds and the producer is never throttled, so a large send (notably sendFile, where the Rust core reads the whole file itself) can queue the entire payload in JVM heap.
Consider bounding the queue and giving the core a way to feel backpressure, e.g. a bounded channel combined with a mechanism that pauses the core's production (or blocks/defers further writes) when the queue is full, rather than dropping packets. At minimum, correct the comment which claims the unbounded channel restores the previous implementation's backpressure.
Was this helpful? React with 👍 or 👎 to provide feedback.
There was a problem hiding this comment.
I left a comment covering this in the swift version: livekit/client-sdk-swift#975 (comment). IMO it's worth reading. Lukas + I (mostly lukas) ended up addressing this same issue on the web via a new "low water mark / high water mark" approach on the data channel which I think if android were to also adopt, it could fix this: livekit/client-sdk-js#2014
IMO fixing this should be out of scope and be a follow up task though, as this pull request is already quite large and this would easily add many hundreds more lines.
| /** Topics we have already warned about, so an unhandled topic logs once rather than per stream. */ | ||
| private val warnedTopics = Collections.synchronizedSet(mutableSetOf<String>()) |
There was a problem hiding this comment.
🟡 Memory grows without bound when a peer sends streams on many unhandled topics
Every topic that arrives without a registered handler is remembered forever in a set that is never pruned (warnedTopics.add(topic) at livekit-android-sdk/src/main/java/io/livekit/android/room/datastream/DataStreams.kt:433), so a peer using many different topic names steadily grows the app's memory.
Impact: A remote participant can make the app's memory use climb indefinitely just by opening streams on unique topic names.
Mechanism
warnedTopics is declared at livekit-android-sdk/src/main/java/io/livekit/android/room/datastream/DataStreams.kt:133 as a plain synchronized mutableSetOf<String>(). warnMissingHandler adds the topic string from an incoming stream header — fully remote-controlled data — and nothing ever removes entries, including across reconnects (abortAllStreams deliberately preserves registration state and does not touch this set).
A bounded structure (e.g. an LRU set with a cap) would keep the once-per-topic logging behaviour while bounding the retained data.
Prompt for agents
DataStreams.warnedTopics is an unbounded synchronized set keyed on the topic string of incoming stream headers, which is attacker/peer controlled, and entries are never removed. Replace it with a bounded structure (for example an LRU-backed set with a modest cap, or clear it on disconnect/abortAllStreams) so that once-per-topic warning behaviour is preserved without letting remote input grow memory indefinitely.
Was this helpful? React with 👍 or 👎 to provide feedback.
| // Forwarded whole rather than destructured: the core re-decodes the packet itself, and v2 | ||
| // headers carry fields (inline content, compression) that only it interprets. The packet has | ||
| // already been decrypted by the engine, which is what the core expects. | ||
| dataStreams.handleIncoming(dp) |
There was a problem hiding this comment.
🟥 End-to-end encryption of incoming data streams is no longer verified
Room.onDataStreamPacket now discards the per-packet encryption type reported by the engine and forwards the raw packet to the core (dataStreams.handleIncoming(dp) at livekit-android-sdk/src/main/java/io/livekit/android/room/Room.kt:1368). The previous implementation compared the packet-level encryption type against the type recorded on the stream header and closed the reader with StreamException.EncryptionTypeMismatch on a mismatch. RTCEngine computes the real value at livekit-android-sdk/src/main/java/io/livekit/android/room/RTCEngine.kt:1340-1347 (NONE unless the packet actually carried an encryptedPacket), and it is now simply dropped. Worse, DataStreams.currentEncryptionType() (livekit-android-sdk/src/main/java/io/livekit/android/room/datastream/DataStreams.kt:313-319) derives StreamInfo.encryptionType from the local room's e2ee configuration, so an unencrypted stream is delivered to application handlers labelled GCM.
Was this helpful? React with 👍 or 👎 to provide feedback.
There was a problem hiding this comment.
I think this is a side effect of the rust core being decoupled from e2ee. Maybe it's worth re-adding this check back in though before the message gets sent into rust?
| /** Topics we have already warned about, so an unhandled topic logs once rather than per stream. */ | ||
| private val warnedTopics = Collections.synchronizedSet(mutableSetOf<String>()) |
There was a problem hiding this comment.
🟨 Unbounded set keyed on remote-supplied stream topics allows memory growth
DataStreams.warnedTopics (livekit-android-sdk/src/main/java/io/livekit/android/room/datastream/DataStreams.kt:133) is a plain synchronized set that accumulates the topic string of every incoming stream that has no registered handler (warnMissingHandler at .../DataStreams.kt:432-439). Topic strings come straight off the wire and are fully controlled by the sending peer, and entries are never removed — not even on disconnect or abortAllStreams.
Was this helpful? React with 👍 or 👎 to provide feedback.
1 participant
Initial port of data streams v2 to the android sdk. The corresponding swift change (which this is fairly heavily patterned off of ) can be found here: livekit/client-sdk-swift#1075
Previously, the android 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.
So, port the android sdk to use the rust sdk data streams v2 implementation, and completely remove the pre-existing kotlin implementation. 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:room.connect( url = wsUrl, token = token, options = ConnectOptions( autoSubscribe = true ), roomOptions = RoomOptions( dynacast = false, dataStream = DataStreamOptions(maxPayloadByteLength = 1000) ) )3. Throwing new data stream errors types
The old kotlin 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.
Adding demo to
sample-appTo exercise these changes, I've added a new section to the
sample-appexample which renders a data stream testing interface. This has been heavily patterned off of the similar interface I added to rust-dev-client here:Uniffi integration
In addition to the data streams v2 features, I've started integrating the uniffi kotlin bindings into this sdk based off of DL's in progress branch. For the most part, this has gone fairly smoothly. I've kept everything wired up with maven local for the time being and will leave it as a cleanup item for an android expert (likely DL) to this working properly prior to merging.
As part of this, I have also fixed two issues in the
livekit-uniffikotlin bindgen:Any uniffi objects which have a method with the name
closedoes not build on uniffi-rs0.31. I made a ticket here: Kotlin bindings cannot have a method namedclose, if it does the generated code isn't valid mozilla/uniffi-rs#2955. I have worked around this by renamingclose->close_streamas a kotlin specific override here: livekit/rust-sdks@1fe0ebaAny uniffi enums which have tagged cases that contain fields names
messagefail to build:Unfortunately, this one can't be fixed in the same way, so I've opted to rename all error cases that contain
messageto instead bereason, also in here: livekit/rust-sdks@1fe0eba. The uniffi-rs maintainers seem to be unable to figure out a good way to fix this - the latest related issue was closed: mozilla/uniffi-rs#2938Warning
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