Data streams v2#1192
Conversation
Changeset ✓This PR includes a changeset covering all affected packages:
|
| 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.
1egoman
force-pushed
the
data-streams-v2
branch
3 times, most recently
from
3bf1532 to
d9e9bb3
Compare
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.
Was this helpful? React with 👍 or 👎 to provide feedback.
There was a problem hiding this comment.
This might be an unrelated bug? My change didn't touch server_version().
1egoman
removed request for
MaxHeimbrock,
alan-george-lk,
lukasIO,
stephen-derosa and
xianshijing-lk
| let inline_content = header.inline_content.take(); | ||
| let is_compressed = header.compression() == proto::CompressionType::DeflateRaw; | ||
|
|
||
| let Ok(info) = AnyStreamInfo::try_from_with_encryption(header, encryption_type.into()) |
There was a problem hiding this comment.
🟡 Receive-side inline detection always reports false under the end-to-end test feature
The inline-content field is stripped from the header (header.inline_content.take() at livekit-data-stream/src/incoming/mod.rs:140) before the stream-info struct is constructed, so the test-only is_inline diagnostic is always false on the receive side.
Impact: The is_inline field on received stream info is always wrong under the __e2e-test feature, making it impossible to assert inline delivery from the receiver's perspective.
Mechanism: inline_content taken before info construction
At livekit-data-stream/src/incoming/mod.rs:140, header.inline_content.take() moves the inline content out of the header into a local variable. The header (now with inline_content = None) is then passed to AnyStreamInfo::try_from_with_encryption at line 143, which calls ByteStreamInfo::from_headers_with_encryption or TextStreamInfo::from_headers_with_encryption.
In those functions (livekit-data-stream/src/info.rs:133 and livekit-data-stream/src/info.rs:162), the __e2e-test feature checks !header.inline_content().is_empty() to set is_inline. Since inline_content was already taken, this always returns an empty slice, so is_inline is always false.
The current e2e tests only check is_inline on the sender's returned StreamInfo (not the receiver's), so no test currently fails. But the receive-side diagnostic is silently broken.
Was this helpful? React with 👍 or 👎 to provide feedback.
Breaks out a subset of the changes from #1192 into a pull request which exclusively moves the existing data streams implementation into its own crate, `livekit-data-streams`, and adds a new `livekit-common` crate (kept fairly minimal for now) for common utilities which `livekit-data-streams` and `livekit` both need. This is largely being scoped to a "lift and shift" type effort, more extensive refactoring of the `livekit-data-streams` crate will be done in #1192. ### Breaking changes There should not be any breaking changes from the perspective of `livekit`'s public api. If so, please surface them. ### Testing This pull request adds the relevant v1 data streams tests from #1192. More test coverage will be introduced as a follow up in the data streams v2 change. --------- Co-authored-by: Jacob Gelman <3182119+ladvoc@users.noreply.github.com>
This comment was marked as outdated.
This comment was marked as outdated.
Sorry, something went wrong.
| /// retaining any trailing incomplete codepoint for the next chunk. | ||
| fn reframe_text(&mut self, decompressed: Vec<u8>) -> Bytes { | ||
| self.pending_text.extend_from_slice(&decompressed); | ||
| let valid = match std::str::from_utf8(&self.pending_text) { |
There was a problem hiding this comment.
any chance that the developer is sending the non-UTF data over ?
There was a problem hiding this comment.
This only is used in the send_text path, and if the data streams v2 protocol is properly being followed, then this should always be proper utf8.
I will think about if there's a good way to catch this being invalid utf8... the part that makes it challenging is that I'm not sure if it's possible to easily distinguish between "a buffer containing all but the final byte of a codepoint" and "a buffer with junk in it". Ideas definitely welcome.
There was a problem hiding this comment.
At least in Rust and Swift, strings are always UTF-8, so there is no possibly of a user sending non-UTF8 data through the public data streams APIs. We should evaluate for other client SDKs and see if the same is true (I'm guessing not in the case of C++).
| self.on_header(header, participant_identity, encryption_type).await | ||
| } | ||
| Packet::Chunk { chunk, encryption_type } => { | ||
| self.on_chunk(chunk, participant_identity, encryption_type).await |
There was a problem hiding this comment.
It looks like it sequentially process the stream chunks, is there any performance concern since on_chunk() will do
match decompressor.push(&chunk.content).await ?
Any pros and cons if we parallelize it thru tokio ? for instance.
There was a problem hiding this comment.
There isn't I don't think a way around this, since decompressor is stateful and must process chunks in order for them to be properly decompressed. Open to ideas though if you can think of something.
There was a problem hiding this comment.
This is why, for data tracks, each track gets its own async task—(de)packetization, (de)compression (once we add support for that), etc. do not block the manager actors from processing new events. However, that is a performance optimization that would require some more extensive changes here to implement, so I would say skip for now.
There was a problem hiding this comment.
Ah, I didn't make the connection shijing meant across multiple concurrently read data streams... yes, good point, I hadn't considered that. I will explore it and if it's fairly easy I will do it now, otherwise I'll hold off as jacob suggested.
| return; | ||
| } | ||
| } | ||
| // A gap is unrecoverable for a stateful decompressor. |
There was a problem hiding this comment.
if it is a long last stream, can it be somehow recoverable ?
There was a problem hiding this comment.
As far as I know, there isn't much that can be done here. Previous compressed chunks may have data that future compressed chunks rely on. Definitely interested in any ideas though if you can think of anything!
Also as a datapoint - data streams use reliable data channels. So if a chunk is missed, that shouldn't be transport related, and instead means it's a bug with the sending client. Currently data streams v1 doesn't tolerate missing chunks currently either so this already is something that clients deal with.
| // Compress once up front when eligible (the deflate work happens at most once, cached in | ||
| // the returned `Vec`), then decide whether it's worth using. | ||
| let use_compression = can_compress | ||
| && maybe_compressed.collect().await.is_ok_and(|c| c.len() < text_bytes.len()); |
There was a problem hiding this comment.
looks like maybe_compressed.collect().await is called multiple times ?
does it make sense to use a variable and collect() once ?
There was a problem hiding this comment.
FYI - collect isn't collecting an iterable. maybe_compressed is a MaybeCollectedAsyncReader (here) and collect() is lazy / only will actually collect the data the first time it is called. The idea here is the first time the compressed data is needed, it will be computed (why .await is needed), and every future time it just returns an immediately resolving future with the data. Storing this in advance means that you potentially compress data sometimes when you don't need to or vice versa which is less efficient.
Arguably .collect() as a method name makes it seem like it's an iterator, maybe changing this name could make this more clear?
…an extra allocation
…s and remove derive Default
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
4 participants
Overview
This change introduces data streams v2, a set of major updates to data streams to add three things:
For information about important decisions made in data streams v2, see the spec. In particular, parts 1-5 provide a good high level overview of data streams v1 + v2, what these protocol improvements are aiming to acheive, and why certain choices were made when designing the protocol.
Also maybe of interest is 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.