Data streams v2#1192
Conversation
ChangesetThe following package versions will be affected by this PR:
|
| bytes = "1.10.1" | ||
| bmrng = "0.5.2" | ||
| flate2 = "1" | ||
| base64 = "0.22" |
There was a problem hiding this comment.
Note that flate2 is now a required dependency of livekit. Previously it was an optional dependency of livekit-api but that's it.
There was a problem hiding this comment.
Not sure if this is is worth doing yet, but it might be worth adding a "compression" feature to this crate to conditionally include this dep. It would be enabled by default, but would provide users who do not want compression to disable it at the feature level.
There was a problem hiding this comment.
I lean towards not doing it and adding it later if people wanted to opt out. But, I could be convinced otherwise if there's a really good use case. Also, in case it's relevant, flate2 is ~80kb.
There was a problem hiding this comment.
My concern would be more how it affects compile time and binary size; probably insignificant but it might be worth checking with cargo build --timings and cargo bloat --release --crates (crate) respectively.
| /// Streams a file from disk to participants as a byte stream. | ||
| /// | ||
| /// Never uses the inline single-packet path (deciding inline-eligibility would require | ||
| /// buffering and compressing the whole file up front). Compresses when every recipient | ||
| /// supports it. The whole file is never buffered in memory at once. |
There was a problem hiding this comment.
Note to self: make this docstring more user centric and not talk about the implementation.
I'll let jacob's in flight pull request push this all the way to the latest
…_compressed / is_inline" This reverts commit 3768723.
Add MaybeCompressed so that compression only ever happens at most once even if it has to happen speculatively sometimes
Not needed for now, can be re-added with an feature later
…ternal -> is_internal
The old version used a derivation of a c algorithm which very few people would have familiarly with. So adjust this to use `rand` instead.
| #[allow(clippy::too_many_arguments)] | ||
| fn text_header( | ||
| id: &str, | ||
| total_length: Option<u64>, | ||
| attributes: HashMap<String, String>, | ||
| inline_content: Option<Vec<u8>>, | ||
| compression: proto::CompressionType, | ||
| ) -> proto::Header { | ||
| proto::Header { | ||
| stream_id: id.to_string(), | ||
| timestamp: 0, | ||
| topic: "topic".to_string(), | ||
| mime_type: "text/plain".to_string(), | ||
| total_length, | ||
| encryption_type: 0, | ||
| attributes, | ||
| content_header: Some(proto::header::ContentHeader::TextHeader( |
There was a problem hiding this comment.
I would be curious if reviewers think these functions help make the test easier to understand or harder, I could go either way.
| impl<T> TakeCell<T> | ||
| where | ||
| T: StreamReader, | ||
| { | ||
| /// Takes the reader out of the cell if its info matches the given predicate. | ||
| /// | ||
| /// Use this method to conditionally handle incoming streams based on info fields | ||
| /// such as topic or attributes. | ||
| /// | ||
| /// This method will only take the reader if the provided predicate returns `true` when called with the reader's info. | ||
| /// If the predicate returns `false` or the reader has already been taken, this method returns `None`. | ||
| /// | ||
| pub fn take_if(&self, predicate: impl FnOnce(&T::Info) -> bool) -> Option<T> { | ||
| self.take_if_raw(|reader| predicate(reader.info())) | ||
| } | ||
| } | ||
|
|
There was a problem hiding this comment.
Note to reviewers - I'm not sure what this TakeCell code was here for. TakeCell::take_if was not being used anywhere in the codebase so I just removed it outright. It compiles and all tests pass without this being here, so I'm worried I may have missed something.
| } | ||
|
|
||
| pub(crate) enum AnyStreamReader { | ||
| pub enum AnyStreamReader { |
There was a problem hiding this comment.
Note to reviewers - some of the visibilities on some of these members had to change now that livekit-data-tracks is exposing previously crate-internal members so they can be pulled in by other crates. From the perspective of a consumer of the livekit crate, all members should have the same visibility as before.
| /// Generates a random stream identifier (UUID v4). | ||
| fn create_random_uuid() -> String { | ||
| uuid::Uuid::new_v4().to_string() | ||
| } |
There was a problem hiding this comment.
Note to reviewers: previously, this used libwebrtc::native::create_random_uuid. I didn't want to make libwebrtc a dependency of this new crate so I swapped this out for uuid::Uuid. I think this should be fine but please surface if there wa s a good reason create_random_uuid was being used instead to generate data stream ids here.
| #[derive(Clone, Default, Debug, Eq, PartialEq)] | ||
| pub struct StreamTextOptions { | ||
| pub topic: String, | ||
| pub attributes: HashMap<String, String>, | ||
| pub destination_identities: Vec<ParticipantIdentity>, | ||
| pub id: Option<String>, | ||
| pub operation_type: Option<OperationType>, | ||
| pub version: Option<i32>, | ||
| pub reply_to_stream_id: Option<String>, | ||
| pub attached_stream_ids: Vec<String>, | ||
| pub generated: Option<bool>, | ||
| /// Whether to deflate-raw compress the payload when all recipients support it. | ||
| /// Defaults to `true` (compression opt-out). Ignored by the incremental `stream_text`. | ||
| pub compress: Option<bool>, | ||
| } |
There was a problem hiding this comment.
Note to reviewers: compress is a new field but otherwise this should be identical. Should I introduce a new builder style construction pattern here and for StreamByteOptions?
| /// A struct which manages the state of data which potentially may need to be compressed in the | ||
| /// future. | ||
| /// | ||
| /// By storing the compressed text optionally after performing compression, we can be sure that co | ||
| /// compression will only ever happen once, even if it must happen as part of speculative paths | ||
| /// (like checking whether compressed bytes are bigger than the literal bytes). | ||
| struct MaybeCompressed<'a> { | ||
| uncompressed: &'a [u8], | ||
| compressed: Option<Vec<u8>>, | ||
| } | ||
|
|
||
| impl<'a> MaybeCompressed<'a> { | ||
| fn new(uncompressed: &'a [u8]) -> Self { | ||
| Self { uncompressed, compressed: None } | ||
| } | ||
|
|
||
| /// Upconverts the Uncompressed variant into the Compressed variant, and returns a reference to | ||
| /// the compressed bytes as a result. | ||
| fn as_compressed(&mut self) -> Result<&[u8], std::io::Error> { | ||
| match &mut self.compressed { | ||
| Some(compressed) => Ok(&*compressed), | ||
| compressed_option @ None => { | ||
| let mut encoder = | ||
| flate2::write::DeflateEncoder::new(Vec::new(), flate2::Compression::default()); | ||
| encoder.write_all(self.uncompressed)?; | ||
| *compressed_option = Some(encoder.finish()?); | ||
| let Some(ref data) = compressed_option else { | ||
| unreachable!("compressed data just set") | ||
| }; | ||
| Ok(data) | ||
| } | ||
| } | ||
| } | ||
| } |
There was a problem hiding this comment.
Note to reviewers: this is pretty cool, this struct ensures that compression will only ever happen once.
| [features] | ||
| # End-to-end testing hooks (exposes is_compressed/is_inline on stream info). Forwarded from | ||
| # the `livekit` crate's `__lk-e2e-test` feature. | ||
| __e2e-test = [] | ||
| # Exposes constructors used by downstream crates' test suites (e.g. `TextStreamReader::new_for_test`). | ||
| test-utils = [] |
There was a problem hiding this comment.
Note to reviewers: should these be separate features or just one?
| /// Take ownership of the value in the cell if it matches some predicate. | ||
| /// | ||
| /// This method will only take the value if the provided predicate returns `true` when called with the current value. | ||
| /// If the predicate returns `false` or the value has already been taken, this method returns `None`. | ||
| pub(crate) fn take_if_raw(&self, predicate: impl FnOnce(&T) -> bool) -> Option<T> { | ||
| if self.value.read().as_ref().map_or(false, |v| predicate(v)) { | ||
| self.take() | ||
| } else { | ||
| None | ||
| } | ||
| } | ||
|
|
There was a problem hiding this comment.
Note to reviewers - TakeCell::take_if_raw wasn't being used at all across the whole codebase, so I removed it along with TakeCell::take_if from my earlier comment.
1egoman
requested review from
MaxHeimbrock,
alan-george-lk,
lukasIO,
stephen-derosa and
xianshijing-lk
as code owners
![devin-ai-integration[bot]](https://avatars.githubusercontent.com/in/811515?s=60&v=4){kind=small}
| self.0.outgoing_stream_manager.send_text(text, options, self.0.as_ref()).await | ||
| } | ||
|
|
||
| fn server_version(&self) -> Option<String> { |
There was a problem hiding this comment.
🔍 Inverted logic in server_version() will skip version checks or panic
The server_version() method at livekit/src/room/rpc/mod.rs:112 uses .and_then(|info| info.version.is_empty().then(|| info.version)) which returns Some(version) only when the version string IS empty, and None when it's non-empty. This is inverted from the intended behavior. When a server reports a valid version like "1.9.0", the method returns None, so the minimum-version check at livekit/src/room/rpc/client.rs:60-66 is skipped entirely. When the server reports an empty version, it returns Some(""), causing Version::parse("").unwrap() to panic. The fix should be (!info.version.is_empty()).then(|| info.version). This is pre-existing (not introduced by this PR) but is in the same impl block that was refactored.
(Refers to lines 105-113)
Was this helpful? React with 👍 or 👎 to provide feedback.
| let content = if is_compressed { | ||
| match inflate_raw(&content) { | ||
| Ok(decompressed) => decompressed, | ||
| Err(error) => { | ||
| // Defensive: a conforming sender never sends a compressed stream we | ||
| // can't read, but drop gracefully if it happens. | ||
| let _ = chunk_tx.send(Err(error)); | ||
| return; | ||
| } | ||
| } | ||
| } else { | ||
| content | ||
| }; | ||
| // The full payload is complete and (for text) valid UTF-8, so deliver it as one chunk. | ||
| if !content.is_empty() { | ||
| let _ = chunk_tx.send(Ok(Bytes::from(content))); | ||
| } | ||
| // Dropping `chunk_tx` closes the reader. | ||
| return; |
There was a problem hiding this comment.
🟨 Unbounded decompression of inline content could allow memory exhaustion
The inflate_raw function (livekit-data-stream/src/incoming/mod.rs:98-104) decompresses inline content with no output size limit via read_to_end. A malicious authenticated participant could send a small compressed payload (fitting within the ~15KB data channel MTU) that decompresses to a very large size, potentially exhausting memory. While the multi-chunk path has a LengthExceeded check against bytes_total, the inline path at lines 166-184 decompresses the entire payload before any size validation.
Was this helpful? React with 👍 or 👎 to provide feedback.
There was a problem hiding this comment.
This is a good point, I wonder if maybe there needs to be some sort of maximum allowed size for a compressed data stream payload which the decompressor enforces. I'd be curious other folks thoughts here...
2 participants
Overview
This change introduces data streams v2, a set of major updates to data streams to add three things:
More info about data streams v2 can be found on the corresponding web sdk pull request. These updates are fully backwards compatible - the sdk will fall back to data streams v1 whenever an old participant is in the room.
There is also a significant amount of new test coverage to data streams (both v1 and v2). There were some small bugs I found along the way which I fixed as well.
Lastly, as part of this, I have broken our data streams into a new
livekit-data-streamscrate (and some common utility data only dependencies (ie, struct types, traits, etc) whichlivekitandlivekit-data-streamsboth need into a newlivekit-commoncrate). This also sets us up well to be able to exposelivekit-data-streamsover uniffi and use it within the context of other clients.