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.
1egoman
force-pushed
the
data-streams-v2
branch
3 times, most recently
from
3bf1532 to
d9e9bb3
Compare
| #[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.
| } | ||
|
|
||
| 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.
| #[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.
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.
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.
Prompt for agents
In handle_header (livekit-data-stream/src/incoming/mod.rs), the inline_content is taken from the header at line 140 before the header is passed to AnyStreamInfo::try_from_with_encryption at line 143. This means the __e2e-test feature's is_inline field (set in info.rs from_headers_with_encryption via header.inline_content().is_empty()) will always be false on the receive side.
To fix this, either:
1. Capture the is_inline boolean before the take (e.g. let is_inline = header.inline_content.is_some()) and pass it through to the info construction, or
2. Restructure so the info is built before the inline_content is taken from the header, or
3. Only take inline_content after building the info struct.
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>
Add MaybeCompressed so that compression only ever happens at most once even if it has to happen speculatively sometimes
… sending a data stream This matches with how web does it
… can_compress / should_compress more clear
…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.
…stry, and CLIENT_PROTOCOL_DATA_STREAM_V2 from livekit-common" This reverts commit ee02f26.
…est-utils" This reverts commit 8e956b4.
…llectedAsyncReader
…ing the protobufs This local layer matches how livekit-datatrack works and gives us space to make local decoupled schema updates without effecting the public types.
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.