livekit_data_stream/outgoing/
manager.rs

1// Copyright 2026 LiveKit, Inc.
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use bmrng::unbounded::{UnboundedRequestReceiver, UnboundedRequestSender};
16use chrono::Utc;
17use livekit_common::{
18    ClientCapability, ParticipantIdentity, RemoteParticipantRegistry,
19    CLIENT_PROTOCOL_DATA_STREAM_V2,
20};
21use livekit_protocol as proto;
22use std::{path::Path, sync::Arc};
23use tokio::sync::Mutex;
24
25use crate::{
26    info::{ByteStreamInfo, TextStreamInfo},
27    types::{ByteHeader, CompressionType, ContentHeader, Header, StreamId, TextHeader},
28    utf8_chunk::Utf8AwareChunkExt,
29    utils::{SendError, StreamError, StreamResult},
30};
31
32use super::{
33    constants,
34    raw_stream::{RawStream, RawStreamOpenOptions},
35    stream_writer::{ByteStreamWriter, TextStreamWriter},
36    StreamByteOptions, StreamTextOptions,
37};
38
39/// Generates a random stream identifier (UUID v4).
40fn create_random_uuid() -> String {
41    uuid::Uuid::new_v4().to_string()
42}
43
44#[derive(Clone)]
45pub struct Manager {
46    /// Request channel for sending packets.
47    packet_tx: UnboundedRequestSender<proto::DataPacket, Result<(), SendError>>,
48}
49
50impl Manager {
51    pub fn new() -> (Self, UnboundedRequestReceiver<proto::DataPacket, Result<(), SendError>>) {
52        let (packet_tx, packet_rx) = bmrng::unbounded_channel();
53        let manager = Self { packet_tx };
54        (manager, packet_rx)
55    }
56
57    pub async fn stream_text(&self, options: StreamTextOptions) -> StreamResult<TextStreamWriter> {
58        // Incremental streams are never inlined or compressed (the content is unknown up front).
59        let stream_id: StreamId = options.id.clone().unwrap_or_else(create_random_uuid).into();
60        let dests = options.destination_identities.clone();
61        let (header, text_header) =
62            build_text_header(&options, stream_id, None, None, CompressionType::None);
63        enforce_header_size(&header, &dests)?;
64
65        let open_options = RawStreamOpenOptions {
66            header: header.clone(),
67            destination_identities: dests,
68            sender_identity: options.sender_identity.clone(),
69            packet_tx: self.packet_tx.clone(),
70        };
71        let writer = TextStreamWriter::new(
72            Arc::new(TextStreamInfo::from_headers(header, text_header)),
73            Arc::new(Mutex::new(RawStream::open(open_options).await?)),
74        );
75        Ok(writer)
76    }
77
78    pub async fn stream_bytes(&self, options: StreamByteOptions) -> StreamResult<ByteStreamWriter> {
79        let stream_id: StreamId = options.id.clone().unwrap_or_else(create_random_uuid).into();
80        let name = options.name.clone().unwrap_or_default();
81        let dests = options.destination_identities.clone();
82        let (header, byte_header) = build_byte_header(
83            &options,
84            stream_id,
85            name,
86            options.total_length,
87            None,
88            CompressionType::None,
89        );
90        enforce_header_size(&header, &dests)?;
91
92        let open_options = RawStreamOpenOptions {
93            header: header.clone(),
94            destination_identities: dests,
95            sender_identity: options.sender_identity.clone(),
96            packet_tx: self.packet_tx.clone(),
97        };
98        let writer = ByteStreamWriter::new(
99            Arc::new(ByteStreamInfo::from_headers(header, byte_header)),
100            Arc::new(Mutex::new(RawStream::open(open_options).await?)),
101        );
102        Ok(writer)
103    }
104
105    pub async fn send_text(
106        &self,
107        text: &str,
108        options: StreamTextOptions,
109        remote_participant_registry: &dyn RemoteParticipantRegistry,
110    ) -> StreamResult<TextStreamInfo> {
111        let stream_id: StreamId = options.id.clone().unwrap_or_else(create_random_uuid).into();
112        let total_length = text.len() as u64;
113
114        let eligibility =
115            evaluate_eligibility(remote_participant_registry, &options.destination_identities);
116        let can_compress = options.compress.unwrap_or(true) && eligibility.compression;
117
118        let text_bytes = text.as_bytes();
119        let mut maybe_compressed = MaybeCollectedAsyncReader::from_async_reader(
120            async_compression::futures::bufread::DeflateEncoder::new(
121                futures_util::io::Cursor::new(text_bytes.to_vec()),
122            ),
123        );
124
125        // Compress once up front when eligible (the deflate work happens at most once, cached in
126        // the returned `Vec`), then decide whether it's worth using.
127        let use_compression = can_compress
128            && maybe_compressed.as_bytes().await.is_ok_and(|c| c.len() < text_bytes.len());
129
130        // 1. Inline single-packet attempt (no attachments; all recipients are >= v2).
131        let (mut header, text_header) = if use_compression {
132            build_text_header(
133                &options,
134                stream_id.clone(),
135                Some(total_length),
136                Some(maybe_compressed.as_bytes().await?.to_owned()),
137                CompressionType::DeflateRaw,
138            )
139        } else {
140            build_text_header(
141                &options,
142                stream_id.clone(),
143                Some(total_length),
144                Some(text_bytes.to_vec()),
145                CompressionType::None,
146            )
147        };
148
149        let proto_header = header.clone().into();
150        if eligibility.inline
151            && options.attached_stream_ids.is_empty()
152            && header_packet_fits(&proto_header, &options.destination_identities)
153        {
154            let mut packet =
155                RawStream::create_header_packet(proto_header, options.destination_identities);
156            packet.participant_identity =
157                options.sender_identity.map(|id| id.into()).unwrap_or_default();
158            RawStream::send_packet(&self.packet_tx, packet).await?;
159            return Ok(TextStreamInfo::from_headers(header, text_header));
160        }
161
162        // 2/3. Chunked, compressed when eligible else uncompressed.
163        header.inline_content = None;
164        enforce_header_size(&header, &options.destination_identities)?;
165
166        let open_options = RawStreamOpenOptions {
167            header: header.clone(),
168            destination_identities: options.destination_identities,
169            sender_identity: options.sender_identity,
170            packet_tx: self.packet_tx.clone(),
171        };
172        let info = TextStreamInfo::from_headers(header, text_header);
173        let mut stream = RawStream::open(open_options).await?;
174        if use_compression {
175            let compressed_bytes = maybe_compressed.as_bytes().await?;
176            stream.write_raw_chunks(compressed_bytes).await?;
177        } else {
178            for chunk in text_bytes.utf8_aware_chunks(constants::STREAM_CHUNK_SIZE_BYTES) {
179                stream.write_chunk(chunk).await?;
180            }
181        }
182        stream.close(None, None).await?;
183        Ok(info)
184    }
185
186    /// Send bytes to participants in the room.
187    ///
188    /// This method sends an in-memory blob of bytes to participants in the room
189    /// as a byte stream. It opens a stream using the provided options, writes the
190    /// entire buffer, and closes the stream before returning.
191    ///
192    /// The `total_length` in the header is set from the provided data and is not
193    /// overridable by `options.total_length`. The header defaults `name` to `"unknown"`
194    /// and `mime_type` to `"application/octet-stream"`.
195    pub async fn send_bytes(
196        &self,
197        data: impl AsRef<[u8]>,
198        options: StreamByteOptions,
199        remote_participant_registry: &dyn RemoteParticipantRegistry,
200    ) -> StreamResult<ByteStreamInfo> {
201        if options.total_length.is_some() {
202            log::warn!("Ignoring total_length option specified for send_bytes");
203        }
204        let bytes = data.as_ref();
205        let stream_id: StreamId = options.id.clone().unwrap_or_else(create_random_uuid).into();
206        let name = options.name.clone().unwrap_or_else(|| constants::BYTE_DEFAULT_NAME.to_owned());
207        let total_length = bytes.len() as u64;
208
209        let eligibility =
210            evaluate_eligibility(remote_participant_registry, &options.destination_identities);
211        let can_compress = options.compress.unwrap_or(true) && eligibility.compression;
212
213        let mut maybe_compressed = MaybeCollectedAsyncReader::from_async_reader(
214            async_compression::futures::bufread::DeflateEncoder::new(
215                futures_util::io::Cursor::new(bytes.to_vec()),
216            ),
217        );
218
219        // Compress once up front when eligible (the deflate work happens at most once, cached in
220        // the returned `Vec`), then decide whether it's worth using.
221        let use_compression =
222            can_compress && maybe_compressed.as_bytes().await.is_ok_and(|c| c.len() < bytes.len());
223
224        // 1. Inline single-packet attempt (if all recipients are >= v2).
225        let (mut header, byte_header) = if use_compression {
226            build_byte_header(
227                &options,
228                stream_id.clone(),
229                name.clone(),
230                Some(total_length), // NOTE: this is purposely always uncompressed length
231                Some(maybe_compressed.as_bytes().await?.to_owned()),
232                CompressionType::DeflateRaw,
233            )
234        } else {
235            build_byte_header(
236                &options,
237                stream_id.clone(),
238                name.clone(),
239                Some(total_length), // NOTE: this is purposely always uncompressed length
240                Some(bytes.to_vec()),
241                CompressionType::None,
242            )
243        };
244
245        let proto_header = header.clone().into();
246        if eligibility.inline && header_packet_fits(&proto_header, &options.destination_identities)
247        {
248            let mut packet =
249                RawStream::create_header_packet(proto_header, options.destination_identities);
250            packet.participant_identity =
251                options.sender_identity.map(|id| id.into()).unwrap_or_default();
252            RawStream::send_packet(&self.packet_tx, packet).await?;
253            return Ok(ByteStreamInfo::from_headers(header, byte_header));
254        }
255
256        // 2/3. Chunked, compressed when eligible else uncompressed.
257        header.inline_content = None;
258        enforce_header_size(&header, &options.destination_identities)?;
259
260        let open_options = RawStreamOpenOptions {
261            header: header.clone(),
262            destination_identities: options.destination_identities,
263            sender_identity: options.sender_identity,
264            packet_tx: self.packet_tx.clone(),
265        };
266        let info = ByteStreamInfo::from_headers(header, byte_header);
267        let mut stream = RawStream::open(open_options).await?;
268        if use_compression {
269            let compressed_bytes = maybe_compressed.as_bytes().await?;
270            stream.write_raw_chunks(compressed_bytes).await?;
271        } else {
272            stream.write_raw_chunks(bytes).await?;
273        }
274        stream.close(None, None).await?;
275        Ok(info)
276    }
277
278    /// Streams a file from disk to participants as a byte stream.
279    ///
280    /// Never uses the inline single-packet path (deciding inline-eligibility would require
281    /// buffering and compressing the whole file up front). Compresses when every recipient
282    /// supports it. The whole file is never buffered in memory at once.
283    pub async fn send_file(
284        &self,
285        path: impl AsRef<Path>,
286        options: StreamByteOptions,
287        remote_participant_registry: &dyn RemoteParticipantRegistry,
288    ) -> StreamResult<ByteStreamInfo> {
289        let path = path.as_ref();
290        let file_size = tokio::fs::metadata(path)
291            .await
292            .map(|metadata| metadata.len())
293            .map_err(StreamError::from)?;
294        let name = options.name.clone().unwrap_or_else(|| {
295            path.file_name().and_then(|n| n.to_str()).unwrap_or_default().to_owned()
296        });
297        let stream_id: StreamId = options.id.clone().unwrap_or_else(create_random_uuid).into();
298        let dests = options.destination_identities.clone();
299
300        let eligibility = evaluate_eligibility(remote_participant_registry, &dests);
301        let should_compress = options.compress.unwrap_or(true) && eligibility.compression;
302        let compression =
303            if should_compress { CompressionType::DeflateRaw } else { CompressionType::None };
304
305        let (header, byte_header) =
306            build_byte_header(&options, stream_id, name, Some(file_size), None, compression);
307        enforce_header_size(&header, &dests)?;
308
309        let open_options = RawStreamOpenOptions {
310            header: header.clone(),
311            destination_identities: dests,
312            sender_identity: options.sender_identity.clone(),
313            packet_tx: self.packet_tx.clone(),
314        };
315        let info = ByteStreamInfo::from_headers(header, byte_header);
316        let mut stream = RawStream::open(open_options).await?;
317        stream.write_file(path, should_compress).await?;
318        stream.close(None, None).await?;
319        Ok(info)
320    }
321}
322
323/// Inline / compression eligibility evaluated over a send's recipients.
324struct SendEligibility {
325    /// Every recipient advertises `clientProtocol >= 2`.
326    inline: bool,
327    /// Inline-eligible AND every recipient advertises `CAP_COMPRESSION_DEFLATE_RAW`.
328    compression: bool,
329}
330
331/// Evaluates inline/compression eligibility over a send's recipients.
332///
333/// Recipients are the named `destinations`, or every remote participant for a broadcast
334/// (empty `destinations`). An empty recipient set (empty room) is eligible for everything.
335fn evaluate_eligibility(
336    registry: &dyn RemoteParticipantRegistry,
337    destinations: &[ParticipantIdentity],
338) -> SendEligibility {
339    let recipients: Vec<ParticipantIdentity> =
340        if destinations.is_empty() { registry.remote_identities() } else { destinations.to_vec() };
341    let inline = recipients
342        .iter()
343        .all(|id| registry.remote_client_protocol(id) >= CLIENT_PROTOCOL_DATA_STREAM_V2);
344    let compression = inline
345        && recipients.iter().all(|id| {
346            registry.remote_capabilities(id).contains(&ClientCapability::CompressionDeflateRaw)
347        });
348
349    SendEligibility { inline, compression }
350}
351
352/// Wraps an [`AsyncRead`] whose bytes are produced lazily (e.g. a deflate encoder), caching them
353/// on first [`Self::collect`] so the underlying work runs at most once. Later `collect` calls
354/// return the cached bytes without re-reading.
355///
356/// (The streaming, still-a-reader-after-collect variant needed by `send_file` will be added when
357/// that path is migrated; today's callers only need `collect`.)
358enum MaybeCollectedAsyncReader<Reader: futures_util::io::AsyncRead + Unpin> {
359    Reader(Reader),
360    Collected(Vec<u8>),
361}
362
363impl<Reader: futures_util::io::AsyncRead + Unpin> MaybeCollectedAsyncReader<Reader> {
364    fn from_async_reader(reader: Reader) -> Self {
365        Self::Reader(reader)
366    }
367
368    async fn as_bytes(&mut self) -> Result<&[u8], std::io::Error> {
369        use futures_util::io::AsyncReadExt;
370        match self {
371            Self::Collected(_) => { /* no-op, handled below */ }
372            Self::Reader(reader) => {
373                let mut buf = Vec::new();
374                reader.read_to_end(&mut buf).await?;
375                *self = Self::Collected(buf);
376            }
377        }
378        let Self::Collected(bytes) = self else { unreachable!("just set to Collected") };
379        Ok(bytes)
380    }
381}
382
383/// Whether the serialized header `DataPacket` fits within the MTU budget.
384fn header_packet_fits(
385    header: &proto::data_stream::Header,
386    destinations: &[ParticipantIdentity],
387) -> bool {
388    use prost::Message;
389    let packet = RawStream::create_header_packet(header.clone(), destinations.to_vec());
390    packet.encoded_len() <= constants::STREAM_CHUNK_SIZE_BYTES
391}
392
393/// Enforces the header-packet MTU budget on the chunked path (the inline path falls back
394/// gracefully instead of erroring).
395fn enforce_header_size(header: &Header, destinations: &[ParticipantIdentity]) -> StreamResult<()> {
396    let proto_header: proto::data_stream::Header = header.clone().into();
397    if header_packet_fits(&proto_header, destinations) {
398        Ok(())
399    } else {
400        Err(StreamError::HeaderTooLarge)
401    }
402}
403
404fn build_text_header(
405    options: &StreamTextOptions,
406    stream_id: StreamId,
407    total_length: Option<u64>,
408    inline_content: Option<Vec<u8>>,
409    compression: CompressionType,
410) -> (Header, TextHeader) {
411    let text_header = TextHeader {
412        operation_type: options.operation_type.unwrap_or_default(),
413        version: options.version.unwrap_or_default(),
414        reply_to_stream_id: options.reply_to_stream_id.clone().map(StreamId::from),
415        attached_stream_ids: options
416            .attached_stream_ids
417            .clone()
418            .into_iter()
419            .map(StreamId::from)
420            .collect(),
421        generated: options.generated.unwrap_or_default(),
422    };
423    let header = Header {
424        stream_id,
425        timestamp: Utc::now().timestamp_millis(),
426        topic: options.topic.clone(),
427        mime_type: constants::TEXT_MIME_TYPE.to_owned(),
428        total_length,
429        attributes: options.attributes.clone(),
430        content_header: Some(ContentHeader::TextHeader(text_header.clone().into())),
431        inline_content,
432        compression,
433    };
434    (header, text_header)
435}
436
437fn build_byte_header(
438    options: &StreamByteOptions,
439    stream_id: StreamId,
440    name: String,
441    total_length: Option<u64>,
442    inline_content: Option<Vec<u8>>,
443    compression: CompressionType,
444) -> (Header, ByteHeader) {
445    let byte_header = ByteHeader { name };
446    let header = Header {
447        stream_id,
448        timestamp: Utc::now().timestamp_millis(),
449        topic: options.topic.clone(),
450        mime_type: options
451            .mime_type
452            .clone()
453            .unwrap_or_else(|| constants::BYTE_MIME_TYPE.to_owned()),
454        total_length,
455        attributes: options.attributes.clone(),
456        content_header: Some(byte_header.clone().into()),
457        inline_content,
458        compression,
459    };
460    (header, byte_header)
461}
462
463#[cfg(test)]
464mod tests {
465    use super::*;
466    use crate::{backend::somewhat_compressible, outgoing::StreamWriter};
467    use livekit_common::{CLIENT_PROTOCOL_DATA_STREAM_RPC, CLIENT_PROTOCOL_DEFAULT};
468    use std::{collections::HashMap, sync::Mutex as StdMutex};
469
470    // --- Fake recipient registry ---------------------------------------------------------
471
472    struct FakeRegistry {
473        remotes: HashMap<String, (i32, Vec<ClientCapability>)>,
474    }
475
476    impl FakeRegistry {
477        fn new() -> Self {
478            Self { remotes: HashMap::new() }
479        }
480
481        fn add(mut self, id: &str, client_protocol: i32, caps: &[ClientCapability]) -> Self {
482            self.remotes.insert(id.to_string(), (client_protocol, caps.to_vec()));
483            self
484        }
485    }
486
487    impl RemoteParticipantRegistry for FakeRegistry {
488        fn remote_client_protocol(&self, identity: &ParticipantIdentity) -> i32 {
489            self.remotes.get(&identity.0).map(|(p, _)| *p).unwrap_or(0)
490        }
491        fn remote_capabilities(&self, identity: &ParticipantIdentity) -> Vec<ClientCapability> {
492            self.remotes.get(&identity.0).map(|(_, c)| c.clone()).unwrap_or_default()
493        }
494        fn remote_identities(&self) -> Vec<ParticipantIdentity> {
495            self.remotes.keys().map(|k| ParticipantIdentity(k.clone())).collect()
496        }
497    }
498
499    fn pre_v2_room() -> FakeRegistry {
500        FakeRegistry::new()
501            .add("alice", CLIENT_PROTOCOL_DEFAULT, &[])
502            .add("bob", CLIENT_PROTOCOL_DEFAULT, &[])
503            .add("jim", CLIENT_PROTOCOL_DATA_STREAM_RPC, &[])
504    }
505
506    fn all_v2_room() -> FakeRegistry {
507        FakeRegistry::new()
508            .add(
509                "alice",
510                CLIENT_PROTOCOL_DATA_STREAM_V2,
511                &[ClientCapability::CompressionDeflateRaw],
512            )
513            .add("bob", CLIENT_PROTOCOL_DATA_STREAM_V2, &[ClientCapability::CompressionDeflateRaw])
514            .add("noCompression", CLIENT_PROTOCOL_DATA_STREAM_V2, &[])
515    }
516
517    fn mixed_room() -> FakeRegistry {
518        FakeRegistry::new()
519            .add("alice", CLIENT_PROTOCOL_DEFAULT, &[])
520            .add("bob", CLIENT_PROTOCOL_DATA_STREAM_V2, &[ClientCapability::CompressionDeflateRaw])
521            .add("jim", CLIENT_PROTOCOL_DATA_STREAM_V2, &[ClientCapability::CompressionDeflateRaw])
522            .add("mallory", CLIENT_PROTOCOL_DEFAULT, &[])
523            .add("noCompression", CLIENT_PROTOCOL_DATA_STREAM_V2, &[])
524    }
525
526    /// A room where every participant is v2 AND advertises the compression capability.
527    fn all_v2_capable_room() -> FakeRegistry {
528        FakeRegistry::new()
529            .add(
530                "alice",
531                CLIENT_PROTOCOL_DATA_STREAM_V2,
532                &[ClientCapability::CompressionDeflateRaw],
533            )
534            .add("bob", CLIENT_PROTOCOL_DATA_STREAM_V2, &[ClientCapability::CompressionDeflateRaw])
535    }
536
537    // --- Capture harness -----------------------------------------------------------------
538
539    type Sent = Arc<StdMutex<Vec<proto::DataPacket>>>;
540
541    fn setup() -> (Manager, Sent) {
542        let (manager, mut packet_rx) = Manager::new();
543        let sent: Sent = Arc::new(StdMutex::new(Vec::new()));
544        let sink = sent.clone();
545        tokio::spawn(async move {
546            while let Ok((packet, responder)) = packet_rx.recv().await {
547                sink.lock().unwrap().push(packet);
548                let _ = responder.respond(Ok(()));
549            }
550        });
551        (manager, sent)
552    }
553
554    fn ids(list: &[&str]) -> Vec<ParticipantIdentity> {
555        list.iter().map(|s| ParticipantIdentity(s.to_string())).collect()
556    }
557
558    fn text_opts(topic: &str, dests: &[&str]) -> StreamTextOptions {
559        StreamTextOptions::new_with_topic(topic).with_destination_identities(ids(dests))
560    }
561
562    fn byte_opts(topic: &str, dests: &[&str]) -> StreamByteOptions {
563        StreamByteOptions::new_with_topic(topic).with_destination_identities(ids(dests))
564    }
565
566    fn header(p: &proto::DataPacket) -> &proto::data_stream::Header {
567        match p.value.as_ref().unwrap() {
568            proto::data_packet::Value::StreamHeader(h) => h,
569            _ => panic!("expected stream header"),
570        }
571    }
572
573    fn chunk(p: &proto::DataPacket) -> &proto::data_stream::Chunk {
574        match p.value.as_ref().unwrap() {
575            proto::data_packet::Value::StreamChunk(c) => c,
576            _ => panic!("expected stream chunk"),
577        }
578    }
579
580    fn is_text_header(h: &proto::data_stream::Header) -> bool {
581        matches!(h.content_header, Some(proto::data_stream::header::ContentHeader::TextHeader(_)))
582    }
583
584    fn is_byte_header(h: &proto::data_stream::Header) -> bool {
585        matches!(h.content_header, Some(proto::data_stream::header::ContentHeader::ByteHeader(_)))
586    }
587
588    fn assert_trailer(p: &proto::DataPacket) {
589        match p.value.as_ref().unwrap() {
590            proto::data_packet::Value::StreamTrailer(t) => assert_eq!(t.reason, ""),
591            _ => panic!("expected stream trailer"),
592        }
593    }
594
595    /// Uniform random bytes — genuinely incompressible (unlike `pseudo_random_text`'s ascii).
596    fn random_bytes(len: usize) -> Vec<u8> {
597        use rand::{rngs::StdRng, Rng, SeedableRng};
598        let mut rng = StdRng::seed_from_u64(0xdead_beef);
599        (0..len).map(|_| rng.random::<u8>()).collect()
600    }
601
602    fn text_content_header(h: &proto::data_stream::Header) -> &proto::data_stream::TextHeader {
603        match h.content_header.as_ref().unwrap() {
604            proto::data_stream::header::ContentHeader::TextHeader(t) => t,
605            _ => panic!("expected text header"),
606        }
607    }
608
609    mod room_with_pre_data_streams_v2_participants {
610        use super::*;
611
612        #[tokio::test]
613        async fn pre_v2_short_text_is_legacy_multipacket() {
614            let (m, sent) = setup();
615            m.send_text("hello world", text_opts("chat", &[]), &pre_v2_room()).await.unwrap();
616            let p = sent.lock().unwrap().clone();
617            assert_eq!(p.len(), 3);
618            let h = header(&p[0]);
619            assert!(is_text_header(h));
620            assert_eq!(h.topic, "chat");
621            assert_eq!(h.compression(), proto::data_stream::CompressionType::None);
622            assert!(h.inline_content.is_none());
623            let c = chunk(&p[1]);
624            assert_eq!(c.chunk_index, 0);
625            assert_eq!(c.content, b"hello world");
626            assert_trailer(&p[2]);
627        }
628
629        #[tokio::test]
630        async fn pre_v2_long_text_splits_at_mtu() {
631            let (m, sent) = setup();
632            let text = "A".repeat(40_000);
633            m.send_text(&text, text_opts("chat", &[]), &pre_v2_room()).await.unwrap();
634            let p = sent.lock().unwrap().clone();
635            assert_eq!(p.len(), 5); // header + 3 chunks + trailer
636            assert_eq!(header(&p[0]).compression(), proto::data_stream::CompressionType::None);
637            assert_eq!(chunk(&p[1]).content.len(), 15_000);
638            assert_eq!(chunk(&p[2]).content.len(), 15_000);
639            assert_eq!(chunk(&p[3]).content.len(), 10_000);
640            assert_eq!(chunk(&p[1]).chunk_index, 0);
641            assert_eq!(chunk(&p[3]).chunk_index, 2);
642            assert_trailer(&p[4]);
643        }
644
645        #[tokio::test]
646        async fn pre_v2_bytes_is_legacy_multipacket() {
647            let (m, sent) = setup();
648            m.send_bytes([0u8, 1, 2, 3], byte_opts("blob", &[]), &pre_v2_room()).await.unwrap();
649            let p = sent.lock().unwrap().clone();
650            assert_eq!(p.len(), 3);
651            let h = header(&p[0]);
652            assert!(is_byte_header(h));
653            assert_eq!(h.compression(), proto::data_stream::CompressionType::None);
654            assert!(h.inline_content.is_none());
655            assert_eq!(chunk(&p[1]).content, vec![0, 1, 2, 3]);
656            assert_trailer(&p[2]);
657        }
658
659        #[tokio::test]
660        async fn pre_v2_empty_text_sends_header_and_trailer() {
661            let (m, sent) = setup();
662            m.send_text("", text_opts("chat", &[]), &pre_v2_room()).await.unwrap();
663            // A well-formed (if contentless) stream: header declaring zero length, then the trailer.
664            let p = sent.lock().unwrap().clone();
665            assert_eq!(p.len(), 2);
666            assert_eq!(header(&p[0]).total_length, Some(0));
667            assert_trailer(&p[1]);
668        }
669    }
670
671    mod room_with_all_data_streams_v2_participants {
672        use super::*;
673
674        mod send_text {
675            use super::*;
676
677            #[tokio::test]
678            async fn v2_short_compressible_text_inlines_compressed() {
679                let (m, sent) = setup();
680                let text = "hello hello compressible world";
681                m.send_text(text, text_opts("chat", &["alice", "bob"]), &all_v2_room())
682                    .await
683                    .unwrap();
684                let p = sent.lock().unwrap().clone();
685                assert_eq!(p.len(), 1);
686                let h = header(&p[0]);
687                assert!(is_text_header(h));
688                assert_eq!(h.compression(), proto::data_stream::CompressionType::DeflateRaw);
689                let inline = h.inline_content.as_ref().unwrap();
690                assert_ne!(inline.as_slice(), text.as_bytes()); // compressed, not raw
691            }
692
693            #[tokio::test]
694            async fn v2_short_incompressible_text_inlines_raw() {
695                let (m, sent) = setup();
696                m.send_text("short", text_opts("chat", &["alice", "bob"]), &all_v2_room())
697                    .await
698                    .unwrap();
699                let p = sent.lock().unwrap().clone();
700                assert_eq!(p.len(), 1);
701                let h = header(&p[0]);
702                assert_eq!(h.compression(), proto::data_stream::CompressionType::None);
703                assert_eq!(h.inline_content.as_ref().unwrap().as_slice(), b"short");
704            }
705
706            #[tokio::test]
707            async fn v2_no_compression_cap_inlines_raw() {
708                let (m, sent) = setup();
709                let text = "hello hello compressible world";
710                m.send_text(text, text_opts("chat", &["noCompression"]), &all_v2_room())
711                    .await
712                    .unwrap();
713                let p = sent.lock().unwrap().clone();
714                assert_eq!(p.len(), 1); // inline (gated on protocol) still happens
715                let h = header(&p[0]);
716                assert_eq!(h.compression(), proto::data_stream::CompressionType::None); // compression gated off by missing cap
717                assert_eq!(h.inline_content.as_ref().unwrap().as_slice(), text.as_bytes());
718            }
719
720            #[tokio::test]
721            async fn v2_large_highly_compressible_text_still_inlines() {
722                let (m, sent) = setup();
723                let text = "hello world".repeat(20_000);
724                m.send_text(&text, text_opts("chat", &["alice", "bob"]), &all_v2_room())
725                    .await
726                    .unwrap();
727                let p = sent.lock().unwrap().clone();
728                assert_eq!(p.len(), 1);
729                let h = header(&p[0]);
730                assert_eq!(h.compression(), proto::data_stream::CompressionType::DeflateRaw);
731                assert!(h.inline_content.as_ref().unwrap().len() < text.len());
732            }
733
734            #[tokio::test]
735            async fn v2_somewhat_compressible_text_is_compressed_multipacket() {
736                let (m, sent) = setup();
737                let text = somewhat_compressible(50_000);
738                m.send_text(&text, text_opts("chat", &["alice", "bob"]), &all_v2_room())
739                    .await
740                    .unwrap();
741                let p = sent.lock().unwrap().clone();
742                let h = header(&p[0]);
743                assert_eq!(h.compression(), proto::data_stream::CompressionType::DeflateRaw);
744                assert!(h.inline_content.is_none());
745                let chunks: Vec<_> = p[1..p.len() - 1].iter().map(chunk).collect();
746                // Multi-packet, but fewer chunks than an uncompressed send would need (ceil(len/15000)).
747                let uncompressed_chunks = text.len().div_ceil(constants::STREAM_CHUNK_SIZE_BYTES);
748                assert!(chunks.len() >= 2);
749                assert!(chunks.len() < uncompressed_chunks);
750                assert_eq!(chunks[0].content.len(), constants::STREAM_CHUNK_SIZE_BYTES); // first chunk is full MTU
751                let total: usize = chunks.iter().map(|c| c.content.len()).sum();
752                assert!(total < text.len()); // compressed
753                assert_trailer(p.last().unwrap());
754            }
755
756            #[tokio::test]
757            async fn v2_compress_false_short_inlines_raw() {
758                let (m, sent) = setup();
759                let text = "hello hello compressible world";
760                let opts = text_opts("chat", &["alice", "bob"]).with_compress(false);
761                m.send_text(text, opts, &all_v2_room()).await.unwrap();
762                let p = sent.lock().unwrap().clone();
763                assert_eq!(p.len(), 1);
764                let h = header(&p[0]);
765                assert_eq!(h.compression(), proto::data_stream::CompressionType::None);
766                assert_eq!(h.inline_content.as_ref().unwrap().as_slice(), text.as_bytes());
767            }
768
769            #[tokio::test]
770            async fn v2_compress_false_large_is_uncompressed_multipacket() {
771                let (m, sent) = setup();
772                let text = "B".repeat(50_000);
773                let opts = text_opts("chat", &["alice", "bob"]).with_compress(false);
774                m.send_text(&text, opts, &all_v2_room()).await.unwrap();
775                let p = sent.lock().unwrap().clone();
776                assert_eq!(p.len(), 6); // header + 4 chunks + trailer
777                assert_eq!(header(&p[0]).compression(), proto::data_stream::CompressionType::None);
778                assert_eq!(chunk(&p[1]).content.len(), 15_000);
779            }
780
781            #[tokio::test]
782            async fn v2_send_text_with_attachments_never_inlines() {
783                let (m, sent) = setup();
784                let opts = text_opts("chat", &["alice", "bob"]).with_attached_stream_id("att1");
785                m.send_text("hello hello compressible world", opts, &all_v2_room()).await.unwrap();
786                let p = sent.lock().unwrap().clone();
787                // Attachments disable the single-packet path even when every recipient is v2.
788                assert_eq!(p.len(), 3); // header + 1 chunk + trailer
789                let h = header(&p[0]);
790                assert!(h.inline_content.is_none());
791                assert_eq!(text_content_header(h).attached_stream_ids, vec!["att1".to_string()]);
792                assert_trailer(&p[2]);
793            }
794
795            #[tokio::test]
796            async fn v2_large_text_to_uncapable_recipient_is_uncompressed_multipacket() {
797                let (m, sent) = setup();
798                let text = "A".repeat(40_000);
799                m.send_text(&text, text_opts("chat", &["noCompression"]), &all_v2_room())
800                    .await
801                    .unwrap();
802                // Inline is attempted (recipient is v2) but the raw payload overflows the MTU, and
803                // compression is gated off by the missing capability — so this falls back to an
804                // uncompressed multi-packet stream split at STREAM_CHUNK_SIZE_BYTES.
805                let p = sent.lock().unwrap().clone();
806                assert_eq!(p.len(), 5);
807                let h = header(&p[0]);
808                assert_eq!(h.compression(), proto::data_stream::CompressionType::None);
809                assert!(h.inline_content.is_none());
810                assert_eq!(chunk(&p[1]).content.len(), 15_000);
811                assert_eq!(chunk(&p[2]).content.len(), 15_000);
812                assert_eq!(chunk(&p[3]).content.len(), 10_000);
813                assert_trailer(&p[4]);
814            }
815
816            #[tokio::test]
817            async fn v2_empty_text_sends_single_inline_packet() {
818                let (m, sent) = setup();
819                m.send_text("", text_opts("chat", &["alice", "bob"]), &all_v2_room())
820                    .await
821                    .unwrap();
822                let p = sent.lock().unwrap().clone();
823                assert_eq!(p.len(), 1);
824                let h = header(&p[0]);
825                // Deflate framing can't shrink an empty payload, so the raw (empty) bytes are kept.
826                assert_eq!(h.compression(), proto::data_stream::CompressionType::None);
827                assert_eq!(h.inline_content.as_deref(), Some(&[][..]));
828                assert_eq!(h.total_length, Some(0));
829            }
830        }
831
832        mod send_bytes {
833            use super::*;
834
835            #[tokio::test]
836            async fn v2_send_bytes_short_incompressible_inlines_raw() {
837                let (m, sent) = setup();
838                m.send_bytes([0u8, 1, 2, 3], byte_opts("blob", &["alice", "bob"]), &all_v2_room())
839                    .await
840                    .unwrap();
841                let p = sent.lock().unwrap().clone();
842                assert_eq!(p.len(), 1);
843                let h = header(&p[0]);
844                // Tiny payload doesn't shrink under DEFLATE framing, so it's sent raw.
845                assert_eq!(h.compression(), proto::data_stream::CompressionType::None);
846                assert_eq!(h.inline_content.as_ref().unwrap().as_slice(), &[0u8, 1, 2, 3]);
847            }
848
849            #[tokio::test]
850            async fn v2_send_bytes_no_compression_cap_inlines_raw() {
851                let (m, sent) = setup();
852                let payload = "hello hello compressible world".as_bytes();
853                m.send_bytes(payload, byte_opts("blob", &["noCompression"]), &all_v2_room())
854                    .await
855                    .unwrap();
856                let p = sent.lock().unwrap().clone();
857                assert_eq!(p.len(), 1); // inline is gated on clientProtocol alone
858                let h = header(&p[0]);
859                assert_eq!(h.compression(), proto::data_stream::CompressionType::None); // compression gated off by the missing cap
860                assert_eq!(h.inline_content.as_ref().unwrap().as_slice(), payload);
861            }
862
863            #[tokio::test]
864            async fn v2_send_bytes_large_highly_compressible_inlines() {
865                let (m, sent) = setup();
866                let payload = vec![0x01u8; 50_000];
867                m.send_bytes(&payload, byte_opts("blob", &["alice", "bob"]), &all_v2_room())
868                    .await
869                    .unwrap();
870                let p = sent.lock().unwrap().clone();
871                assert_eq!(p.len(), 1); // compresses well under the MTU, so it still goes inline
872                let h = header(&p[0]);
873                assert_eq!(h.compression(), proto::data_stream::CompressionType::DeflateRaw);
874                assert!(h.inline_content.as_ref().unwrap().len() < payload.len());
875                assert_eq!(h.total_length, Some(50_000)); // always the PRE-compression length
876            }
877
878            #[tokio::test]
879            async fn v2_send_bytes_somewhat_compressible_is_compressed_multipacket() {
880                let (m, sent) = setup();
881                let payload = somewhat_compressible(50_000).into_bytes();
882                m.send_bytes(&payload, byte_opts("blob", &["alice", "bob"]), &all_v2_room())
883                    .await
884                    .unwrap();
885                let p = sent.lock().unwrap().clone();
886                let h = header(&p[0]);
887                assert!(is_byte_header(h));
888                assert_eq!(h.compression(), proto::data_stream::CompressionType::DeflateRaw);
889                assert!(h.inline_content.is_none());
890                let chunks: Vec<_> = p[1..p.len() - 1].iter().map(chunk).collect();
891                assert!(chunks.len() >= 2);
892                assert!(chunks.len() < payload.len().div_ceil(constants::STREAM_CHUNK_SIZE_BYTES));
893                assert_eq!(chunks[0].content.len(), constants::STREAM_CHUNK_SIZE_BYTES);
894                assert_trailer(p.last().unwrap());
895            }
896
897            #[tokio::test]
898            async fn v2_send_bytes_compress_false_large_is_uncompressed_multipacket() {
899                let (m, sent) = setup();
900                let payload = vec![0x07u8; 40_000];
901                let opts = byte_opts("blob", &["alice", "bob"]).with_compress(false);
902                m.send_bytes(&payload, opts, &all_v2_room()).await.unwrap();
903                let p = sent.lock().unwrap().clone();
904                assert_eq!(p.len(), 5); // header + 15k/15k/10k chunks + trailer
905                let h = header(&p[0]);
906                assert_eq!(h.compression(), proto::data_stream::CompressionType::None);
907                assert!(h.inline_content.is_none());
908                assert_eq!(chunk(&p[1]).content.len(), 15_000);
909                assert_eq!(chunk(&p[2]).content.len(), 15_000);
910                assert_eq!(chunk(&p[3]).content.len(), 10_000);
911                assert!(chunk(&p[3]).content.iter().all(|b| *b == 0x07));
912                assert_trailer(&p[4]);
913            }
914
915            #[tokio::test]
916            async fn v2_send_bytes_short_compressible_inlines_compressed() {
917                let (m, sent) = setup();
918                let payload = "hello hello compressible world".as_bytes().to_vec();
919                let mut opts = byte_opts("blob", &["alice", "bob"]);
920                opts.attributes.insert("foo".to_string(), "bar".to_string());
921                let info = m.send_bytes(&payload, opts, &all_v2_room()).await.unwrap();
922                let p = sent.lock().unwrap().clone();
923                assert_eq!(p.len(), 1);
924                let h = header(&p[0]);
925                assert!(is_byte_header(h));
926                assert_eq!(h.compression(), proto::data_stream::CompressionType::DeflateRaw);
927                assert_ne!(h.inline_content.as_ref().unwrap().as_slice(), payload.as_slice());
928                assert_eq!(info.name, "unknown");
929                assert_eq!(info.mime_type, "application/octet-stream");
930                assert_eq!(info.total_length, Some(payload.len() as u64));
931                assert_eq!(info.attributes().get("foo"), Some(&"bar".to_string()));
932            }
933        }
934
935        mod send_file {
936            use super::*;
937
938            async fn write_temp_file(bytes: &[u8]) -> std::path::PathBuf {
939                let path =
940                    std::env::temp_dir().join(format!("lk_ds_test_{}.bin", create_random_uuid()));
941                tokio::fs::write(&path, bytes).await.unwrap();
942                path
943            }
944
945            #[tokio::test]
946            async fn send_file_never_inlines_and_compresses_when_eligible() {
947                let (m, sent) = setup();
948                let path = write_temp_file(&vec![0x01u8; 10_000]).await;
949                m.send_file(&path, byte_opts("file", &["alice", "bob"]), &all_v2_room())
950                    .await
951                    .unwrap();
952                let _ = tokio::fs::remove_file(&path).await;
953                let p = sent.lock().unwrap().clone();
954                assert_eq!(p.len(), 3); // header + 1 chunk + trailer, NOT inline
955                let h = header(&p[0]);
956                assert!(is_byte_header(h));
957                assert_eq!(h.compression(), proto::data_stream::CompressionType::DeflateRaw);
958                assert!(h.inline_content.is_none());
959                assert!(chunk(&p[1]).content.len() < 10_000); // compressed
960                assert_trailer(&p[2]);
961            }
962
963            #[tokio::test]
964            async fn send_file_uncompressed_splits_at_mtu() {
965                let (m, sent) = setup();
966                let path = write_temp_file(&vec![0x07u8; 20_000]).await;
967                m.send_file(&path, byte_opts("file", &[]).with_compress(false), &all_v2_room())
968                    .await
969                    .unwrap();
970                let _ = tokio::fs::remove_file(&path).await;
971                let p = sent.lock().unwrap().clone();
972                assert_eq!(p.len(), 4); // header + 15000 + 5000 + trailer
973                assert_eq!(header(&p[0]).compression(), proto::data_stream::CompressionType::None);
974                assert_eq!(chunk(&p[1]).content.len(), 15_000);
975                assert_eq!(chunk(&p[2]).content.len(), 5_000);
976                assert_eq!(chunk(&p[2]).chunk_index, 1);
977                assert_trailer(&p[3]);
978            }
979
980            #[tokio::test]
981            async fn send_file_incompressible_compressed_expands() {
982                let (m, sent) = setup();
983                let data = random_bytes(50_000);
984                let path = write_temp_file(&data).await;
985                m.send_file(&path, byte_opts("file", &["alice", "bob"]), &all_v2_room())
986                    .await
987                    .unwrap();
988                let _ = tokio::fs::remove_file(&path).await;
989
990                let p = sent.lock().unwrap().clone();
991                assert_eq!(p.len(), 6); // header + 4 chunks + trailer
992                let h = header(&p[0]);
993                assert_eq!(h.compression(), proto::data_stream::CompressionType::DeflateRaw);
994                assert_eq!(chunk(&p[1]).content.len(), 15_000);
995                // Deflate adds slight overhead on incompressible data — the accepted trade-off for
996                // streaming the file instead of buffering it to make an inline/compression decision.
997                let total: usize =
998                    p[1..p.len() - 1].iter().map(|packet| chunk(packet).content.len()).sum();
999                assert!(total > data.len());
1000                assert_trailer(p.last().unwrap());
1001            }
1002
1003            #[tokio::test]
1004            async fn send_file_to_no_compression_recipient_is_uncompressed() {
1005                let (m, sent) = setup();
1006                let path = write_temp_file(&vec![0x07u8; 10_000]).await;
1007                m.send_file(&path, byte_opts("file", &["noCompression"]), &all_v2_room())
1008                    .await
1009                    .unwrap();
1010                let _ = tokio::fs::remove_file(&path).await;
1011
1012                // v2 recipient without the compression capability in a v2 room: uncompressed
1013                // multi-packet (and never inline).
1014                let p = sent.lock().unwrap().clone();
1015                assert_eq!(p.len(), 3);
1016                let h = header(&p[0]);
1017                assert!(is_byte_header(h));
1018                assert_eq!(h.compression(), proto::data_stream::CompressionType::None);
1019                assert!(h.inline_content.is_none());
1020                let c = chunk(&p[1]);
1021                assert_eq!(c.content.len(), 10_000);
1022                assert!(c.content.iter().all(|b| *b == 0x07));
1023                assert_trailer(&p[2]);
1024            }
1025
1026            #[tokio::test]
1027            async fn send_file_empty_file() {
1028                let (m, sent) = setup();
1029                let path = write_temp_file(&[]).await;
1030                m.send_file(&path, byte_opts("file", &["alice", "bob"]), &all_v2_room())
1031                    .await
1032                    .unwrap();
1033                let _ = tokio::fs::remove_file(&path).await;
1034
1035                // An empty file still produces a well-formed compressed stream: a header declaring zero
1036                // length, the deflate stream's final block, and a trailer.
1037                let p = sent.lock().unwrap().clone();
1038                let h = header(&p[0]);
1039                assert!(is_byte_header(h));
1040                assert_eq!(h.total_length, Some(0));
1041                assert_eq!(h.compression(), proto::data_stream::CompressionType::DeflateRaw);
1042                assert_trailer(p.last().unwrap());
1043            }
1044        }
1045
1046        #[tokio::test]
1047        async fn v2_broadcast_all_capable_inlines_compressed() {
1048            let (m, sent) = setup();
1049            // No destinations: eligibility is evaluated over every remote participant.
1050            m.send_text(
1051                "hello hello compressible world",
1052                text_opts("chat", &[]),
1053                &all_v2_capable_room(),
1054            )
1055            .await
1056            .unwrap();
1057            let p = sent.lock().unwrap().clone();
1058            assert_eq!(p.len(), 1);
1059            assert_eq!(
1060                header(&p[0]).compression(),
1061                proto::data_stream::CompressionType::DeflateRaw
1062            );
1063        }
1064
1065        #[tokio::test]
1066        async fn v2_broadcast_with_uncapable_member_inlines_raw() {
1067            let (m, sent) = setup();
1068            let text = "hello hello compressible world";
1069            // all_v2_room includes "noCompression": inline still applies, compression does not.
1070            m.send_text(text, text_opts("chat", &[]), &all_v2_room()).await.unwrap();
1071            let p = sent.lock().unwrap().clone();
1072            assert_eq!(p.len(), 1);
1073            let h = header(&p[0]);
1074            assert_eq!(h.compression(), proto::data_stream::CompressionType::None);
1075            assert_eq!(h.inline_content.as_ref().unwrap().as_slice(), text.as_bytes());
1076        }
1077
1078        #[tokio::test]
1079        async fn empty_room_broadcast_is_v2_eligible() {
1080            let (m, sent) = setup();
1081            // Nobody to receive it, so nothing gates the v2 fast path.
1082            m.send_text(
1083                "hello hello compressible world",
1084                text_opts("chat", &[]),
1085                &FakeRegistry::new(),
1086            )
1087            .await
1088            .unwrap();
1089            let p = sent.lock().unwrap().clone();
1090            assert_eq!(p.len(), 1);
1091            assert_eq!(
1092                header(&p[0]).compression(),
1093                proto::data_stream::CompressionType::DeflateRaw
1094            );
1095        }
1096    }
1097
1098    mod room_with_mixed_participants {
1099        use super::*;
1100
1101        #[tokio::test]
1102        async fn mixed_broadcast_falls_back_to_legacy() {
1103            let (m, sent) = setup();
1104            m.send_text("hello world", text_opts("chat", &[]), &mixed_room()).await.unwrap();
1105            let p = sent.lock().unwrap().clone();
1106            assert_eq!(p.len(), 3);
1107            assert_eq!(header(&p[0]).compression(), proto::data_stream::CompressionType::None);
1108            assert!(header(&p[0]).inline_content.is_none());
1109            assert_eq!(chunk(&p[1]).content, b"hello world");
1110        }
1111
1112        #[tokio::test]
1113        async fn mixed_targeted_v2_subset_inlines_compressed() {
1114            let (m, sent) = setup();
1115            let text = "hello hello compressible world";
1116            m.send_text(text, text_opts("chat", &["bob", "jim"]), &mixed_room()).await.unwrap();
1117            let p = sent.lock().unwrap().clone();
1118            assert_eq!(p.len(), 1);
1119            let h = header(&p[0]);
1120            assert_eq!(h.compression(), proto::data_stream::CompressionType::DeflateRaw);
1121            assert_ne!(h.inline_content.as_ref().unwrap().as_slice(), text.as_bytes());
1122        }
1123
1124        #[tokio::test]
1125        async fn mixed_targeted_subset_missing_cap_inlines_uncompressed() {
1126            let (m, sent) = setup();
1127            let text = "hello hello compressible world";
1128            m.send_text(text, text_opts("chat", &["bob", "jim", "noCompression"]), &mixed_room())
1129                .await
1130                .unwrap();
1131            let p = sent.lock().unwrap().clone();
1132            assert_eq!(p.len(), 1);
1133            let h = header(&p[0]);
1134            assert_eq!(h.compression(), proto::data_stream::CompressionType::None);
1135            assert_eq!(h.inline_content.as_ref().unwrap().as_slice(), text.as_bytes());
1136        }
1137    }
1138
1139    // --- Sender identity ------------------------------------------------------------------
1140
1141    #[tokio::test]
1142    async fn stream_text_with_sender_identity_stamps_every_packet() {
1143        let (m, sent) = setup();
1144        let opts = text_opts("chat", &[]).with_sender_identity("impostor");
1145        let writer = m.stream_text(opts).await.unwrap();
1146        writer.write("hello").await.unwrap();
1147        writer.close().await.unwrap();
1148        let p = sent.lock().unwrap().clone();
1149        assert_eq!(p.len(), 3);
1150        assert!(p.iter().all(|pkt| pkt.participant_identity == "impostor"));
1151    }
1152
1153    #[tokio::test]
1154    async fn send_text_inline_with_sender_identity_stamps_packet() {
1155        let (m, sent) = setup();
1156        let opts = text_opts("chat", &["alice", "bob"]).with_sender_identity("impostor");
1157        m.send_text("hello hello compressible world", opts, &all_v2_room()).await.unwrap();
1158        let p = sent.lock().unwrap().clone();
1159        assert_eq!(p.len(), 1);
1160        assert_eq!(p[0].participant_identity, "impostor");
1161    }
1162
1163    #[tokio::test]
1164    async fn packets_carry_no_identity_when_sender_identity_unset() {
1165        let (m, sent) = setup();
1166        let writer = m.stream_text(text_opts("chat", &[])).await.unwrap();
1167        writer.close().await.unwrap();
1168        let p = sent.lock().unwrap().clone();
1169        assert!(p.iter().all(|pkt| pkt.participant_identity.is_empty()));
1170    }
1171
1172    // --- Incremental writers never compress or inline ------------------------------------
1173
1174    #[tokio::test]
1175    async fn stream_text_never_compresses_or_inlines() {
1176        let (m, sent) = setup();
1177        let writer = m.stream_text(text_opts("chat", &["noCompression"])).await.unwrap();
1178        assert_eq!(sent.lock().unwrap().len(), 1);
1179        let h0 = sent.lock().unwrap()[0].clone();
1180        assert!(is_text_header(header(&h0)));
1181        assert_eq!(header(&h0).compression(), proto::data_stream::CompressionType::None);
1182        assert!(header(&h0).inline_content.is_none());
1183
1184        writer.write("hello world").await.unwrap();
1185        assert_eq!(sent.lock().unwrap().len(), 2);
1186        assert_eq!(chunk(&sent.lock().unwrap()[1]).content, b"hello world");
1187
1188        writer.close().await.unwrap();
1189        let p = sent.lock().unwrap().clone();
1190        assert_eq!(p.len(), 3);
1191        assert_trailer(&p[2]);
1192    }
1193
1194    #[tokio::test]
1195    async fn stream_bytes_never_compresses_or_inlines() {
1196        let (m, sent) = setup();
1197        let writer = m.stream_bytes(byte_opts("blob", &["noCompression"])).await.unwrap();
1198        assert_eq!(sent.lock().unwrap().len(), 1);
1199        assert_eq!(
1200            header(&sent.lock().unwrap()[0]).compression(),
1201            proto::data_stream::CompressionType::None
1202        );
1203
1204        writer.write(&[0u8, 1, 2, 3]).await.unwrap();
1205        assert_eq!(chunk(&sent.lock().unwrap()[1]).content, vec![0, 1, 2, 3]);
1206
1207        writer.close().await.unwrap();
1208        let p = sent.lock().unwrap().clone();
1209        assert_eq!(p.len(), 3);
1210        assert_trailer(&p[2]);
1211    }
1212
1213    mod header_size_limit {
1214        use super::*;
1215
1216        #[tokio::test]
1217        async fn oversized_attributes_on_chunked_path_errors() {
1218            let (m, _sent) = setup();
1219            let mut opts = text_opts("chat", &[]); // pre-v2 below => chunked path
1220            opts.attributes.insert("big".to_string(), "x".repeat(20_000));
1221            let result = m.send_text("hello", opts, &pre_v2_room()).await;
1222            assert!(matches!(result, Err(StreamError::HeaderTooLarge)));
1223        }
1224
1225        /// Builds a minimal text header carrying the given attributes.
1226        fn header_with_attributes(attributes: HashMap<String, String>) -> Header {
1227            Header {
1228                stream_id: "s1".into(),
1229                timestamp: 0,
1230                topic: "chat".to_string(),
1231                mime_type: constants::TEXT_MIME_TYPE.to_owned(),
1232                total_length: None,
1233                attributes,
1234                content_header: Some(ContentHeader::TextHeader(TextHeader::default())),
1235                inline_content: None,
1236                compression: CompressionType::None,
1237            }
1238        }
1239
1240        #[test]
1241        fn enforce_header_size_accepts_small_header() {
1242            // A minimal header serializes well under the MTU budget, so it is accepted.
1243            let header = header_with_attributes(HashMap::new());
1244            assert!(enforce_header_size(&header, &[]).is_ok());
1245        }
1246
1247        #[test]
1248        fn enforce_header_size_rejects_large_header() {
1249            // A giant attribute value pushes the serialized header past the MTU budget.
1250            let mut attributes = HashMap::new();
1251            attributes.insert("big".to_string(), "x".repeat(20_000));
1252            let header = header_with_attributes(attributes);
1253            assert!(matches!(enforce_header_size(&header, &[]), Err(StreamError::HeaderTooLarge)));
1254        }
1255    }
1256
1257    // Regression test for CLT-2773: dropping a `RawStream` on a thread that has
1258    // no Tokio runtime in TLS (e.g. the .NET GC finalizer thread in the Unity
1259    // SDK) used to panic because `Drop` called `tokio::spawn` unconditionally.
1260    #[test]
1261    fn drop_raw_stream_on_non_tokio_thread_does_not_panic() {
1262        let rt = tokio::runtime::Runtime::new().unwrap();
1263
1264        let raw_stream = rt.block_on(async {
1265            let (packet_tx, mut packet_rx) =
1266                bmrng::unbounded_channel::<proto::DataPacket, Result<(), SendError>>();
1267
1268            tokio::spawn(async move {
1269                while let Ok((_packet, responder)) = packet_rx.recv().await {
1270                    let _ = responder.respond(Ok(()));
1271                }
1272            });
1273
1274            let header = Header {
1275                stream_id: "gc-test-stream".into(),
1276                timestamp: 0,
1277                topic: "gc-test-topic".to_string(),
1278                mime_type: constants::TEXT_MIME_TYPE.to_owned(),
1279                total_length: None,
1280                attributes: HashMap::new(),
1281                content_header: None,
1282                // Data streams v2 fields
1283                inline_content: None,
1284                compression: CompressionType::None,
1285            };
1286
1287            RawStream::open(RawStreamOpenOptions {
1288                header,
1289                destination_identities: vec![],
1290                sender_identity: None,
1291                packet_tx,
1292            })
1293            .await
1294            .expect("RawStream should open")
1295        });
1296
1297        let drop_thread = std::thread::spawn(move || drop(raw_stream));
1298
1299        drop_thread.join().expect("Dropping RawStream on a non-Tokio thread must not panic");
1300    }
1301
1302    // --- Additional spec-conformance cases ------------------------------------------------
1303
1304    mod stream_text_bytes {
1305        use super::*;
1306
1307        #[tokio::test]
1308        async fn stream_bytes_multi_write_splits_at_mtu() {
1309            let (m, sent) = setup();
1310            let writer = m.stream_bytes(byte_opts("blob", &[])).await.unwrap();
1311            writer.write(&vec![0x01u8; 20_000]).await.unwrap();
1312            writer.write(&vec![0x01u8; 20_000]).await.unwrap();
1313            writer.close().await.unwrap();
1314
1315            // Each write splits into a 15k + 5k chunk; indices are contiguous ACROSS writes.
1316            let p = sent.lock().unwrap().clone();
1317            assert_eq!(p.len(), 6); // header + 4 chunks + trailer
1318            for (i, expected_len) in [15_000usize, 5_000, 15_000, 5_000].iter().enumerate() {
1319                let c = chunk(&p[i + 1]);
1320                assert_eq!(c.chunk_index, i as u64);
1321                assert_eq!(c.content.len(), *expected_len);
1322                assert!(c.content.iter().all(|b| *b == 0x01));
1323            }
1324            assert_trailer(&p[5]);
1325        }
1326
1327        #[tokio::test]
1328        async fn close_with_options_sends_trailer_attributes() {
1329            let (m, sent) = setup();
1330            let writer = m.stream_text(text_opts("chat", &[])).await.unwrap();
1331            writer.write("hello").await.unwrap();
1332            let attributes = HashMap::from([("result".to_string(), "ok".to_string())]);
1333            writer.close_with_options(None, Some(attributes.clone())).await.unwrap();
1334
1335            let p = sent.lock().unwrap().clone();
1336            let trailer = match p.last().unwrap().value.as_ref().unwrap() {
1337                proto::data_packet::Value::StreamTrailer(t) => t,
1338                _ => panic!("expected stream trailer"),
1339            };
1340            assert_eq!(trailer.reason, "");
1341            assert_eq!(trailer.attributes, attributes);
1342        }
1343
1344        #[tokio::test]
1345        async fn close_with_options_sends_reason_and_attributes() {
1346            let (m, sent) = setup();
1347            let writer = m.stream_text(text_opts("chat", &[])).await.unwrap();
1348            let attributes = HashMap::from([("cause".to_string(), "cancelled".to_string())]);
1349            writer.close_with_options(Some("aborted"), Some(attributes.clone())).await.unwrap();
1350
1351            let p = sent.lock().unwrap().clone();
1352            let trailer = match p.last().unwrap().value.as_ref().unwrap() {
1353                proto::data_packet::Value::StreamTrailer(t) => t,
1354                _ => panic!("expected stream trailer"),
1355            };
1356            assert_eq!(trailer.reason, "aborted");
1357            assert_eq!(trailer.attributes, attributes);
1358        }
1359
1360        #[tokio::test]
1361        async fn stream_text_oversized_attributes_errors() {
1362            let (m, sent) = setup();
1363            let mut opts = text_opts("chat", &[]);
1364            opts.attributes.insert("big".to_string(), "x".repeat(20_000));
1365            let result = m.stream_text(opts).await;
1366            assert!(matches!(result, Err(StreamError::HeaderTooLarge)));
1367            // The error must be raised before anything hits the wire.
1368            assert!(sent.lock().unwrap().is_empty());
1369        }
1370
1371        #[tokio::test]
1372        async fn stream_text_splits_on_utf8_boundaries_at_mtu() {
1373            let (m, sent) = setup();
1374            // 14 999 single-byte chars put the 4-byte emoji straddling the 15 000-byte MTU boundary.
1375            let text = format!("{}😀{}", "a".repeat(14_999), "b".repeat(10));
1376            let writer = m.stream_text(text_opts("chat", &[])).await.unwrap();
1377            writer.write(&text).await.unwrap();
1378            writer.close().await.unwrap();
1379
1380            let p = sent.lock().unwrap().clone();
1381            assert_eq!(p.len(), 4); // header + 2 chunks + trailer
1382            let first = chunk(&p[1]);
1383            let second = chunk(&p[2]);
1384            // The split backs off below the MTU rather than bisecting the emoji, so each chunk
1385            // decodes independently.
1386            assert_eq!(first.content.len(), 14_999);
1387            let first_str =
1388                std::str::from_utf8(&first.content).expect("chunk 0 must be valid UTF-8");
1389            let second_str =
1390                std::str::from_utf8(&second.content).expect("chunk 1 must be valid UTF-8");
1391            assert_eq!(format!("{first_str}{second_str}"), text);
1392        }
1393    }
1394}
livekit_data_stream/outgoing/manager.rs

livekit_data_stream/outgoing/
manager.rs
