1use crate::client::ClientConfig;
4use crate::quic::CongestionControl;
5use crate::quic::Resolved;
6use crate::quic::ServerId;
7use crate::server::ServerConfig;
8use crate::tls::{FingerprintVerifier, ServeCerts};
9use std::net;
10use std::sync::Arc;
11use std::time::Duration;
12use url::Url;
13
14pub use web_transport_quinn;
15
16fn apply_qlog(transport: &mut quinn::TransportConfig, quic: &Resolved, role: &str) -> Result<()> {
22 let Some(dir) = quic.qlog_dir() else {
25 return Ok(());
26 };
27
28 #[cfg(feature = "qlog")]
29 {
30 let path = dir.join(format!("moq-{role}-{}.sqlog", std::process::id()));
32 let file = std::fs::File::create(&path).map_err(Error::CreateQlog)?;
33
34 let mut config = quinn::QlogConfig::default();
39 config.writer(Box::new(file)).title(Some(format!("moq-native {role}")));
40
41 transport.qlog_stream(config.into_stream());
42 tracing::info!(path = %path.display(), "writing qlog");
43 }
44
45 #[cfg(not(feature = "qlog"))]
46 let _ = (transport, dir, role);
47
48 Ok(())
49}
50
51fn apply_transport(transport: &mut quinn::TransportConfig, quic: &Resolved) {
53 transport.max_idle_timeout(Some(quic.idle_timeout.try_into().expect("idle timeout out of range")));
54 transport.keep_alive_interval(quic.keep_alive);
55
56 if !quic.mtu_discovery {
58 transport.mtu_discovery_config(None);
59 }
60
61 let max_streams = quinn::VarInt::from_u64(quic.max_streams).unwrap_or(quinn::VarInt::MAX);
62 transport.max_concurrent_bidi_streams(max_streams);
63 transport.max_concurrent_uni_streams(max_streams);
64
65 if let Some(gso) = quic.gso {
67 transport.enable_segmentation_offload(gso);
68 }
69
70 transport.congestion_controller_factory(congestion_factory(congestion_control(quic)));
71}
72
73fn congestion_control(quic: &Resolved) -> CongestionControl {
78 quic.congestion_control.unwrap_or(CongestionControl::Delay)
79}
80
81fn congestion_factory(family: CongestionControl) -> Arc<dyn quinn::congestion::ControllerFactory + Send + Sync> {
83 match family {
84 CongestionControl::Loss => Arc::new(quinn::congestion::CubicConfig::default()),
85 CongestionControl::Delay => Arc::new(quinn::congestion::BbrConfig::default()),
86 }
87}
88
89#[derive(Debug, thiserror::Error)]
91#[non_exhaustive]
92pub enum Error {
93 #[error("failed to bind UDP socket")]
95 BindSocket(#[source] std::io::Error),
96
97 #[error("failed to create QUIC endpoint")]
99 CreateEndpoint(#[source] std::io::Error),
100
101 #[error("failed to create qlog file")]
103 CreateQlog(#[source] std::io::Error),
104
105 #[error("no async runtime")]
107 NoRuntime,
108
109 #[error("failed to get local address")]
111 LocalAddr(#[source] std::io::Error),
112
113 #[error("failed to resolve bind address")]
115 ResolveBind(#[source] std::io::Error),
116
117 #[error("invalid DNS name")]
119 InvalidDnsName,
120
121 #[error("failed DNS lookup")]
123 DnsLookup(#[source] std::io::Error),
124
125 #[error("no DNS entries")]
127 NoDnsEntries,
128
129 #[error("failed to fetch fingerprint")]
131 FetchFingerprint(#[source] reqwest::Error),
132
133 #[error("fingerprint request failed")]
135 FingerprintStatus(#[source] reqwest::Error),
136
137 #[error("failed to read fingerprint")]
139 ReadFingerprint(#[source] reqwest::Error),
140
141 #[error("invalid fingerprint")]
143 InvalidFingerprint(#[from] hex::FromHexError),
144
145 #[error("url scheme must be 'https', 'moqt', or 'moql'")]
147 InvalidScheme,
148
149 #[error("unsupported URL scheme: {0}")]
151 UnsupportedScheme(String),
152
153 #[error("missing handshake data")]
155 MissingHandshake,
156
157 #[error("missing ALPN")]
159 MissingAlpn,
160
161 #[error("failed to decode ALPN")]
163 DecodeAlpn(#[from] std::string::FromUtf8Error),
164
165 #[error("unsupported ALPN: {0}")]
167 UnsupportedAlpn(String),
168
169 #[error("missing server name for raw QUIC connection")]
171 MissingServerName,
172
173 #[error("failed to construct URL from server name")]
175 BuildUrl(#[source] url::ParseError),
176
177 #[error("quic_lb_nonce must be at least 4")]
179 QuicLbNonceTooSmall,
180
181 #[error("connection ID length ({0}) exceeds maximum of 20")]
183 QuicLbCidTooLong(usize),
184
185 #[error("failed to build client certificate verifier")]
187 ClientVerifier(#[source] rustls::server::VerifierBuilderError),
188
189 #[error(transparent)]
191 NoInitialCipherSuite(#[from] quinn::crypto::rustls::NoInitialCipherSuite),
192
193 #[error(transparent)]
195 Connect(#[from] quinn::ConnectError),
196
197 #[error(transparent)]
199 Connection(#[from] quinn::ConnectionError),
200
201 #[error(transparent)]
203 Client(#[from] web_transport_quinn::ClientError),
204
205 #[error(transparent)]
207 ConnectRejected(#[from] crate::ConnectError),
208
209 #[error(transparent)]
211 Server(#[from] web_transport_quinn::ServerError),
212
213 #[error("failed to establish QUIC connection")]
215 Establish(#[source] quinn::ConnectionError),
216
217 #[error("failed to receive WebTransport request")]
219 RecvRequest(#[source] web_transport_quinn::ServerError),
220
221 #[error(transparent)]
223 Tls(#[from] crate::tls::Error),
224
225 #[error("all {} connection attempts failed: {}", .0.len(), crate::failover::describe(.0))]
231 Failover(Vec<crate::failover::Failure<Error>>),
232}
233
234impl crate::failover::Aggregate for Error {
235 fn aggregate(failures: Vec<crate::failover::Failure<Self>>) -> Self {
236 Self::Failover(failures)
237 }
238}
239
240type Result<T> = std::result::Result<T, Error>;
241
242#[derive(Clone)]
245pub(crate) struct QuinnClient {
246 pub quic: quinn::Endpoint,
247 pub transport: Arc<quinn::TransportConfig>,
248 pub http_bootstrap: bool,
250 pub host_name: Option<String>,
252 pub failover_delay: Duration,
254 dual_stack: bool,
258}
259
260impl QuinnClient {
261 pub fn new(config: &ClientConfig) -> Result<Self> {
262 let socket = crate::bind::udp(config.bind).map_err(Error::BindSocket)?;
263 let dual_stack = crate::bind::udp_is_dual_stack(&socket);
264
265 let quic = config.quic.resolve();
266 let mut transport = quinn::TransportConfig::default();
267 apply_transport(&mut transport, &quic);
268 apply_qlog(&mut transport, &quic, "client")?;
269 let transport = Arc::new(transport);
270
271 let runtime = quinn::default_runtime().ok_or(Error::NoRuntime)?;
273 let endpoint_config = quinn::EndpointConfig::default();
274
275 let quic = quinn::Endpoint::new(endpoint_config, None, socket, runtime).map_err(Error::CreateEndpoint)?;
277
278 Ok(Self {
279 quic,
280 transport,
281 http_bootstrap: config.tls.allows_http_bootstrap(),
282 host_name: config.tls.host_name.clone(),
283 failover_delay: config.resolved_failover_delay(),
284 dual_stack,
285 })
286 }
287
288 pub async fn connect(
289 &self,
290 tls: &rustls::ClientConfig,
291 url: Url,
292 versions: &moq_net::Versions,
293 ) -> Result<web_transport_quinn::Session> {
294 let mut url = url;
295 let mut config = tls.clone();
296
297 let host = url.host().ok_or(Error::InvalidDnsName)?.to_string();
298 let port = url.port().unwrap_or(443);
299
300 let local = self.quic.local_addr().map_err(Error::LocalAddr)?;
304 let addrs = tokio::net::lookup_host((host.clone(), port))
305 .await
306 .map_err(Error::DnsLookup)?;
307 let candidates = crate::failover::match_local(addrs, local, self.dual_stack);
308 if candidates.is_empty() {
309 return Err(Error::NoDnsEntries);
310 }
311
312 if url.scheme() == "http" {
313 if self.http_bootstrap {
317 let mut fingerprint = url.clone();
319 fingerprint.set_path("/certificate.sha256");
320 fingerprint.set_query(None);
321 fingerprint.set_fragment(None);
322
323 tracing::warn!(url = %fingerprint, "performing insecure HTTP request for certificate");
324
325 let resp = reqwest::get(fingerprint.as_str())
326 .await
327 .map_err(Error::FetchFingerprint)?
328 .error_for_status()
329 .map_err(Error::FingerprintStatus)?;
330
331 let fingerprint = resp.text().await.map_err(Error::ReadFingerprint)?;
332 let fingerprint = hex::decode(fingerprint.trim())?;
333
334 let verifier = FingerprintVerifier::new(config.crypto_provider().clone(), vec![fingerprint]);
335 config.dangerous().set_certificate_verifier(Arc::new(verifier));
336 } else {
337 tracing::warn!(
338 "ignoring insecure http:// fingerprint bootstrap; using the configured TLS verification"
339 );
340 }
341
342 url.set_scheme("https").expect("failed to set scheme");
343 }
344
345 let alpns: Vec<Vec<u8>> = match url.scheme() {
346 "https" => vec![web_transport_quinn::ALPN.as_bytes().to_vec()],
347 "moqt" | "moql" => versions.alpns().iter().map(|alpn| alpn.as_bytes().to_vec()).collect(),
348 _ => return Err(Error::InvalidScheme),
349 };
350
351 config.alpn_protocols = alpns;
352 config.key_log = Arc::new(rustls::KeyLogFile::new());
353
354 let config: quinn::crypto::rustls::QuicClientConfig = config.try_into()?;
355 let mut config = quinn::ClientConfig::new(Arc::new(config));
356 config.transport_config(self.transport.clone());
357
358 tracing::debug!(%url, ?candidates, "connecting");
359
360 let host_name = self.host_name.clone().unwrap_or(host);
362
363 let connection = crate::failover::race(candidates, self.failover_delay, |addr| {
367 let endpoint = self.quic.clone();
368 let config = config.clone();
369 let host_name = host_name.clone();
370 async move { Ok::<_, Error>(endpoint.connect_with(config, addr, &host_name)?.await?) }
371 })
372 .await?;
373 tracing::Span::current().record("id", connection.stable_id());
374
375 let mut request = web_transport_quinn::proto::ConnectRequest::new(url.clone());
376 for alpn in versions.alpns() {
377 request = request.with_protocol(alpn.to_string());
378 }
379
380 let session = match url.scheme() {
381 "https" => web_transport_quinn::Session::connect(connection, request)
382 .await
383 .map_err(map_client_error)?,
384 "moqt" | "moql" => {
385 let handshake = connection
386 .handshake_data()
387 .ok_or(Error::MissingHandshake)?
388 .downcast::<quinn::crypto::rustls::HandshakeData>()
389 .unwrap();
390
391 let alpn = handshake.protocol.ok_or(Error::MissingAlpn)?;
392 let alpn = String::from_utf8(alpn)?;
393
394 let response = web_transport_quinn::proto::ConnectResponse::OK.with_protocol(alpn);
395 web_transport_quinn::Session::raw(connection, request, response)
396 }
397 _ => return Err(Error::UnsupportedScheme(url.scheme().to_string())),
398 };
399
400 Ok(session)
401 }
402}
403
404impl Error {
405 pub(crate) fn connect_error(&self) -> Option<crate::ConnectError> {
406 match self {
407 Self::ConnectRejected(err) => Some(*err),
408 Self::Client(err) => classify_client_error(err),
409 Self::Failover(failures) => failures.iter().find_map(|failure| failure.error.connect_error()),
410 _ => None,
411 }
412 }
413
414 pub(crate) fn status(&self) -> Option<u16> {
419 match self {
420 Self::FetchFingerprint(err) | Self::FingerprintStatus(err) | Self::ReadFingerprint(err) => {
421 err.status().map(|status| status.as_u16())
422 }
423 Self::Client(err) => client_status(err),
424 Self::Failover(failures) => {
428 let mut settled = None;
429 for failure in failures {
430 match failure.error.status() {
431 Some(status) if !crate::error::status_retryable(status) => settled = Some(status),
432 _ => return None,
433 }
434 }
435 settled
436 }
437 _ => None,
438 }
439 }
440}
441
442fn map_client_error(err: web_transport_quinn::ClientError) -> Error {
443 if let Some(err) = classify_client_error(&err) {
444 return err.into();
445 }
446
447 err.into()
448}
449
450fn classify_client_error(err: &web_transport_quinn::ClientError) -> Option<crate::ConnectError> {
451 client_status(err).and_then(crate::ConnectError::from_status_u16)
452}
453
454fn client_status(err: &web_transport_quinn::ClientError) -> Option<u16> {
462 match err {
463 web_transport_quinn::ClientError::HttpError(err) => connect_status(err),
464 _ => None,
465 }
466}
467
468fn connect_status(err: &web_transport_quinn::ConnectError) -> Option<u16> {
469 match err {
470 web_transport_quinn::ConnectError::ErrorStatus(status) => Some(status.as_u16()),
471 web_transport_quinn::ConnectError::ProtoError(err) => proto_status(err),
472 _ => None,
473 }
474}
475
476fn proto_status(err: &web_transport_quinn::proto::ConnectError) -> Option<u16> {
477 match err {
478 web_transport_quinn::proto::ConnectError::ErrorStatus(status)
479 | web_transport_quinn::proto::ConnectError::WrongStatus(Some(status)) => Some(status.as_u16()),
480 _ => None,
481 }
482}
483
484pub(crate) struct QuinnServer {
487 pub quic: quinn::Endpoint,
488 pub certs: Arc<ServeCerts>,
489}
490
491impl QuinnServer {
492 pub fn new(config: ServerConfig) -> Result<Self> {
493 let quic = config.quic.resolve();
494 let mut transport = quinn::TransportConfig::default();
495 apply_transport(&mut transport, &quic);
496 apply_qlog(&mut transport, &quic, "server")?;
497 let transport = Arc::new(transport);
498
499 let provider = crate::crypto::provider();
500
501 let certs = ServeCerts::new(provider.clone());
502 certs.load_certs(&config.tls)?;
503 let certs = Arc::new(certs);
504
505 let tls_builder = rustls::ServerConfig::builder_with_provider(provider.clone())
506 .with_protocol_versions(&[&rustls::version::TLS13])
507 .map_err(crate::tls::Error::from)?;
508
509 let mut tls = if config.tls.root.is_empty() {
510 tls_builder.with_no_client_auth().with_cert_resolver(certs.clone())
511 } else {
512 let roots = config.tls.load_roots()?;
513 let verifier = rustls::server::WebPkiClientVerifier::builder_with_provider(Arc::new(roots), provider)
514 .allow_unauthenticated()
515 .build()
516 .map_err(Error::ClientVerifier)?;
517 tls_builder
518 .with_client_cert_verifier(verifier)
519 .with_cert_resolver(certs.clone())
520 };
521
522 let mut alpns: Vec<Vec<u8>> = config
524 .versions()
525 .alpns()
526 .iter()
527 .map(|alpn| alpn.as_bytes().to_vec())
528 .collect();
529 alpns.push(web_transport_quinn::ALPN.as_bytes().to_vec());
530
531 tls.alpn_protocols = alpns;
532 tls.key_log = Arc::new(rustls::KeyLogFile::new());
533
534 let tls: quinn::crypto::rustls::QuicServerConfig = tls.try_into()?;
535 let mut tls = quinn::ServerConfig::with_crypto(Arc::new(tls));
536 tls.transport_config(transport);
537
538 if let Some(addr) = config.quic.preferred_v4 {
541 tls.preferred_address_v4(Some(addr));
542 }
543 if let Some(addr) = config.quic.preferred_v6 {
544 tls.preferred_address_v6(Some(addr));
545 }
546
547 let runtime = quinn::default_runtime().ok_or(Error::NoRuntime)?;
549
550 let listen =
551 crate::util::resolve(config.bind.as_deref(), crate::server::DEFAULT_BIND).map_err(Error::ResolveBind)?;
552
553 let mut endpoint_config = quinn::EndpointConfig::default();
555 if let Some(server_id) = config.quic.quic_lb_id {
556 let nonce_len = config.quic.quic_lb_nonce.unwrap_or(8);
557 if nonce_len < 4 {
558 return Err(Error::QuicLbNonceTooSmall);
559 }
560
561 let cid_len = 1 + server_id.len() + nonce_len;
562 if cid_len > 20 {
563 return Err(Error::QuicLbCidTooLong(cid_len));
564 }
565
566 tracing::info!(
567 ?server_id,
568 nonce_len,
569 "using QUIC-LB compatible connection ID generation"
570 );
571 endpoint_config.cid_generator(move || Box::new(ServerIdGenerator::new(server_id.clone(), nonce_len)));
572 }
573
574 let socket = crate::bind::udp(listen).map_err(Error::BindSocket)?;
575
576 let quic = quinn::Endpoint::new(endpoint_config, Some(tls), socket, runtime).map_err(Error::CreateEndpoint)?;
578
579 tokio::spawn(crate::tls::reload_certs(certs.clone(), config.tls.clone()));
582
583 Ok(Self { quic, certs })
584 }
585
586 pub fn accept(&self) -> impl std::future::Future<Output = Option<quinn::Incoming>> + '_ {
587 self.quic.accept()
588 }
589
590 pub fn certificates(&self) -> crate::tls::Certificates {
591 crate::tls::Certificates::new(self.certs.info.clone())
592 }
593
594 pub fn local_addr(&self) -> Result<net::SocketAddr> {
595 self.quic.local_addr().map_err(Error::LocalAddr)
596 }
597
598 pub fn close(&self) {
599 self.quic.close(quinn::VarInt::from_u32(0), b"server shutdown");
600 }
601}
602
603pub(crate) async fn accept(
610 conn: quinn::Incoming,
611 alpns: Vec<&'static str>,
612) -> Result<(
613 web_transport_quinn::Session,
614 Option<Url>,
615 Option<crate::tls::PeerIdentity>,
616)> {
617 let mut conn = conn.accept()?;
618
619 let handshake = conn
620 .handshake_data()
621 .await?
622 .downcast::<quinn::crypto::rustls::HandshakeData>()
623 .unwrap();
624
625 let alpn = handshake.protocol.ok_or(Error::MissingAlpn)?;
626 let alpn = String::from_utf8(alpn)?;
627 let host = handshake.server_name.unwrap_or_default();
628
629 tracing::debug!(%host, ip = %conn.remote_address(), %alpn, "accepting");
630
631 let conn = conn.await.map_err(Error::Establish)?;
633
634 let span = tracing::Span::current();
635 span.record("id", conn.stable_id()); tracing::debug!(%host, ip = %conn.remote_address(), %alpn, "accepted");
637
638 match alpn.as_str() {
639 web_transport_quinn::ALPN => {
640 let request = web_transport_quinn::Request::accept(conn)
643 .await
644 .map_err(Error::RecvRequest)?;
645 let url = Some(request.url.clone());
646 let identity = crate::tls::PeerIdentity::from_any(request.conn().peer_identity());
647
648 let mut response = web_transport_quinn::proto::ConnectResponse::OK;
649 if let Some(protocol) = request.protocols.iter().find(|p| alpns.contains(&p.as_str())) {
655 response = response.with_protocol(protocol);
656 }
657 let session = request.respond(response).await.map_err(Error::Server)?;
658 Ok((session, url, identity))
659 }
660 alpn if alpns.contains(&alpn) => {
665 let host_str = if host.contains(':') {
671 format!("[{}]", host)
672 } else {
673 host.clone()
674 };
675 let url = format!("moqt://{}", host_str).parse::<Url>().map_err(Error::BuildUrl)?;
676 let request = web_transport_quinn::proto::ConnectRequest::new(url);
677 let response = web_transport_quinn::proto::ConnectResponse::OK.with_protocol(alpn);
678 let identity = crate::tls::PeerIdentity::from_any(conn.peer_identity());
679 let session = web_transport_quinn::Session::raw(conn, request, response);
681 Ok((session, None, identity))
682 }
683 _ => Err(Error::UnsupportedAlpn(alpn)),
684 }
685}
686
687struct ServerIdGenerator {
690 server_id: ServerId,
691 nonce_len: usize,
692}
693
694impl ServerIdGenerator {
695 fn new(server_id: ServerId, nonce_len: usize) -> Self {
696 Self { server_id, nonce_len }
697 }
698}
699
700impl quinn::ConnectionIdGenerator for ServerIdGenerator {
701 fn generate_cid(&mut self) -> quinn::ConnectionId {
702 use rand::RngExt;
703 let cid_len = self.cid_len();
704 let mut cid = Vec::with_capacity(cid_len);
705 cid.push((cid_len - 1) as u8);
707 cid.extend(self.server_id.0.iter());
708 cid.extend(rand::rng().random_iter::<u8>().take(self.nonce_len));
709 quinn::ConnectionId::new(cid.as_slice())
710 }
711
712 fn cid_len(&self) -> usize {
713 1 + self.server_id.len() + self.nonce_len
714 }
715
716 fn cid_lifetime(&self) -> Option<Duration> {
717 None
718 }
719}
720
721#[cfg(test)]
722mod tests {
723 use super::*;
724
725 fn connect_rejected(status: u16) -> Error {
726 Error::Client(web_transport_quinn::ClientError::HttpError(
727 web_transport_quinn::ConnectError::ErrorStatus(
728 web_transport_quinn::http::StatusCode::from_u16(status).unwrap(),
729 ),
730 ))
731 }
732
733 #[test]
737 fn a_rejected_connect_reports_its_status() {
738 for status in [400, 404, 405, 410, 501] {
739 assert_eq!(connect_rejected(status).status(), Some(status));
740 assert!(
741 !crate::error::status_retryable(status),
742 "{status} should stop the reconnect loop"
743 );
744 }
745
746 for status in [408, 429, 502, 503, 504] {
747 assert_eq!(connect_rejected(status).status(), Some(status));
748 assert!(crate::error::status_retryable(status), "{status} should be retried");
749 }
750
751 assert_eq!(
753 connect_rejected(401).connect_error(),
754 Some(crate::ConnectError::Unauthorized)
755 );
756 assert_eq!(
757 connect_rejected(403).connect_error(),
758 Some(crate::ConnectError::Forbidden)
759 );
760 }
761
762 #[test]
765 fn congestion_factory_maps_each_family() {
766 let now = std::time::Instant::now();
767 let mtu = 1200;
768
769 let loss = congestion_factory(CongestionControl::Loss).build(now, mtu);
770 assert!(loss.into_any().downcast::<quinn::congestion::Cubic>().is_ok());
771
772 let delay = congestion_factory(CongestionControl::Delay).build(now, mtu);
773 assert!(delay.into_any().downcast::<quinn::congestion::Bbr>().is_ok());
774 }
775
776 #[test]
778 fn congestion_control_defaults_to_delay() {
779 let mut quic = crate::quic::Client::default();
780 assert_eq!(congestion_control(&quic.resolve()), CongestionControl::Delay);
781
782 quic.congestion_control = Some(CongestionControl::Loss);
784 assert_eq!(congestion_control(&quic.resolve()), CongestionControl::Loss);
785 }
786
787 #[tokio::test]
790 async fn delay_reaches_the_live_connection() {
791 let server_config = ServerConfig {
792 bind: Some("127.0.0.1:0".to_string()),
793 tls: crate::tls::Server {
794 generate: vec!["localhost".into()],
795 ..Default::default()
796 },
797 quic: crate::quic::Server {
798 congestion_control: Some(CongestionControl::Delay),
799 ..Default::default()
800 },
801 ..Default::default()
802 };
803
804 let server = QuinnServer::new(server_config).expect("server init");
805 let addr = server.local_addr().expect("local addr");
806
807 let accepted = tokio::spawn(async move {
808 let incoming = server.accept().await.expect("no incoming connection");
809 let conn = incoming.accept().expect("accept").await.expect("handshake");
810 conn.congestion_state()
811 .into_any()
812 .downcast::<quinn::congestion::Bbr>()
813 .is_ok()
814 });
815
816 let mut tls_config = crate::tls::Client::default();
818 tls_config.disable_verify = Some(true);
819
820 let client_config = ClientConfig {
821 bind: "127.0.0.1:0".parse().unwrap(),
822 tls: tls_config,
823 quic: crate::quic::Client {
824 congestion_control: Some(CongestionControl::Delay),
825 ..Default::default()
826 },
827 ..Default::default()
828 };
829
830 let tls = client_config.tls.build().expect("tls config");
831 let client = QuinnClient::new(&client_config).expect("client init");
832 let url: Url = format!("moqt://127.0.0.1:{}", addr.port()).parse().unwrap();
834
835 tokio::time::timeout(Duration::from_secs(5), async move {
838 let session = client
839 .connect(&tls, url, &moq_net::Versions::default())
840 .await
841 .expect("connect failed");
842
843 assert!(
845 session
846 .congestion_state()
847 .into_any()
848 .downcast::<quinn::congestion::Bbr>()
849 .is_ok(),
850 "client connection is not running BBR"
851 );
852 assert!(
853 accepted.await.expect("server task panicked"),
854 "server connection is not running BBR"
855 );
856 })
857 .await
858 .expect("test timed out");
859 }
860}