diff --git a/crates/id_effect_axum/src/server/h3.rs b/crates/id_effect_axum/src/server/h3.rs index f28ebe41..5c6546cf 100644 --- a/crates/id_effect_axum/src/server/h3.rs +++ b/crates/id_effect_axum/src/server/h3.rs @@ -639,6 +639,8 @@ impl http_body::Body for H3Body { #[derive(Debug)] pub(crate) struct HostHttp3 { endpoint: QuicEndpoint, + /// Registered at bind, so the host accepts HTTP/3 as soon as it has started. + route: AlpnRoute, } impl HostHttp3 { @@ -665,7 +667,8 @@ impl HostHttp3 { }; let config = QuicEndpointConfig::new(identity.clone()).with_limits(quic_limits(limits)); let endpoint = QuicEndpoint::bind(tcp, config).map_err(io_error)?; - Ok(Some(Self { endpoint })) + let route = endpoint.register(Route::h3()).map_err(io_error)?; + Ok(Some(Self { endpoint, route })) } /// The UDP port (the TCP one). @@ -673,7 +676,8 @@ impl HostHttp3 { self.endpoint.local_addr().port() } - /// [`serve_h3`] with the default [`Http3Options`], then closes the endpoint. + /// [`serve_h3`] on the bound route with the default [`Http3Options`], then closes the + /// endpoint. pub(crate) async fn serve( self, app: Router, @@ -683,8 +687,8 @@ impl HostHttp3 { where F: Future + Send + 'static, { - let served = serve_h3( - &self.endpoint, + let served = serve_route( + Ok(self.route), app, limits, Http3Options::default(), diff --git a/crates/id_effect_axum/src/server/webtransport.rs b/crates/id_effect_axum/src/server/webtransport.rs index 0de37eae..454e3c04 100644 --- a/crates/id_effect_axum/src/server/webtransport.rs +++ b/crates/id_effect_axum/src/server/webtransport.rs @@ -1248,11 +1248,13 @@ mod tests { let app = app(config); let options = Http3Options::default().webtransport(); assert!(options.webtransport_enabled()); + // Called before the spawn, so the route is registered by the time a test connects. + let served = serve_h3(&quic, app, limits, options, async { + let _ = stopped.await; + }); let done = tokio::spawn(async move { - serve_h3(&quic, app, limits, options, async { - let _ = stopped.await; - }) - .await + let _quic = quic; + served.await }); let tls = client_tls_config(&identity, None, &[pin]).unwrap(); Served { diff --git a/crates/id_effect_axum/tests/http3.rs b/crates/id_effect_axum/tests/http3.rs index 0d4f0b4b..b75c6d1e 100644 --- a/crates/id_effect_axum/tests/http3.rs +++ b/crates/id_effect_axum/tests/http3.rs @@ -867,36 +867,15 @@ async fn serve_h3_ends_when_its_endpoint_closes() { QuicEndpointConfig::new(ca.issue().identity()), ) .unwrap(); - let server = { - let endpoint = endpoint.clone(); - tokio::spawn(async move { - serve_h3( - &endpoint, - plain_app(), - ConnectionLimits::default(), - Http3Options::new(), - std::future::pending(), - ) - .await - }) - }; - // `serve_h3` registers the `h3` route when the spawned task first runs; a handshake that - // lands before that finds no route ("peer doesn't support any known protocol", seen on a - // loaded CI runner), so connect until the route is there. - let deadline = tokio::time::Instant::now() + WAIT; - let mut h3 = loop { - match h3_connect(endpoint.local_addr(), ca.roots()).await { - Ok(h3) => break h3, - Err(e) if e.contains("no application protocol") || e.contains("known protocol") => { - assert!( - tokio::time::Instant::now() < deadline, - "h3 route never registered: {e}" - ); - tokio::time::sleep(Duration::from_millis(20)).await; - } - Err(e) => panic!("{e}"), - } - }; + // `serve_h3` registers the `h3` route when called, so a client may connect right away. + let server = tokio::spawn(serve_h3( + &endpoint, + plain_app(), + ConnectionLimits::default(), + Http3Options::new(), + std::future::pending(), + )); + let mut h3 = h3_connect(endpoint.local_addr(), ca.roots()).await.unwrap(); assert_eq!(h3.get("/hello").await.unwrap().body, "hello"); endpoint.close(); tokio::time::timeout(WAIT, server) @@ -931,15 +910,15 @@ async fn serve_h3_on_an_endpoint_of_your_own() { ) .unwrap(); let (stop, stopped) = tokio::sync::oneshot::channel::<()>(); - let server = { - let endpoint = endpoint.clone(); - tokio::spawn(async move { - serve_h3(&endpoint, plain_app(), limits, Http3Options::new(), async { - let _ = stopped.await; - }) - .await - }) - }; + let server = tokio::spawn(serve_h3( + &endpoint, + plain_app(), + limits, + Http3Options::new(), + async { + let _ = stopped.await; + }, + )); let mut h3 = h3_connect(endpoint.local_addr(), ca.roots()).await.unwrap(); assert_eq!(h3.get("/hello").await.unwrap().body, "hello"); // The endpoint routes h3 once: a second server on it is refused. diff --git a/crates/id_effect_mail/tests/mailpit.rs b/crates/id_effect_mail/tests/mailpit.rs index cc77c527..e59f68af 100644 --- a/crates/id_effect_mail/tests/mailpit.rs +++ b/crates/id_effect_mail/tests/mailpit.rs @@ -43,35 +43,48 @@ async fn api(client: &reqwest::Client, mailpit: &MailpitContainer, path: &str) - async fn checks(mailpit: MailpitContainer) { let templates = Templates::from_dir(PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("tests/templates")).unwrap(); - let message = templates - .message( - "welcome", - &json!({ - "user": { "name": "Carol & Co" }, - "confirm_url": "https://ledger.example/confirm", - "company": "Ledger GmbH", - }), - ) - .unwrap() - .from(Mailbox::named( - "Ledger", - Address::parse("hello@ledger.example").unwrap(), - )) - .to(Address::parse("carol@ledger.example").unwrap()) - .build() - .unwrap(); + let message = || { + templates + .message( + "welcome", + &json!({ + "user": { "name": "Carol & Co" }, + "confirm_url": "https://ledger.example/confirm", + "company": "Ledger GmbH", + }), + ) + .unwrap() + .from(Mailbox::named( + "Ledger", + Address::parse("hello@ledger.example").unwrap(), + )) + .to(Address::parse("carol@ledger.example").unwrap()) + .build() + .unwrap() + }; let config = SmtpConfig::plaintext(mailpit.host(), mailpit.smtp_port()); - let sent = - run(deliver::(message).provide_layer(SmtpMailer::layer::(config))) - .await - .expect("Mailpit accepts the message"); + // Mailpit logs readiness before its SMTP listener accepts, and Docker's port proxy accepts + // for it meanwhile ("incomplete response", seen on CI): retry transport failures briefly. + let deadline = Instant::now() + Duration::from_secs(10); + let sent = loop { + let mailer = SmtpMailer::layer::(config.clone()); + match run(deliver::(message()).provide_layer(mailer)).await { + Ok(sent) => break sent, + Err(MailError::Transport(e)) if Instant::now() < deadline => { + eprintln!("Mailpit not accepting yet: {e}"); + tokio::time::sleep(Duration::from_millis(200)).await; + } + Err(e) => panic!("Mailpit accepts the message: {e:?}"), + } + }; let message_id = sent.id.expect("a Message-ID"); let client = reqwest::Client::new(); let deadline = Instant::now() + Duration::from_secs(10); let listed = loop { let list = api(&client, &mailpit, "/api/v1/messages").await; - if list["total"].as_u64() == Some(1) { + // A retried send may have landed twice; Mailpit lists the newest (the one `sent` names) first. + if list["total"].as_u64().is_some_and(|n| n >= 1) { break list; } assert!(Instant::now() < deadline, "no message in Mailpit: {list}"); diff --git a/crates/id_effect_rpc/tests/browser.rs b/crates/id_effect_rpc/tests/browser.rs index 590919eb..ef65934a 100644 --- a/crates/id_effect_rpc/tests/browser.rs +++ b/crates/id_effect_rpc/tests/browser.rs @@ -233,9 +233,17 @@ async fn serve_rpc() -> (SocketAddr, SocketAddr, String) { ) .unwrap(); let udp = quic.local_addr(); + // Called before the spawn, so the route is registered by the time the browser connects. + let served = serve_h3( + &quic, + app, + limits, + Http3Options::default().webtransport(), + std::future::pending(), + ); tokio::spawn(async move { - let options = Http3Options::default().webtransport(); - serve_h3(&quic, app, limits, options, std::future::pending()).await + let _quic = quic; + served.await }); (tcp, udp, hash) } diff --git a/crates/id_effect_rpc/tests/spans.rs b/crates/id_effect_rpc/tests/spans.rs index 37b76a36..d6ac58ab 100644 --- a/crates/id_effect_rpc/tests/spans.rs +++ b/crates/id_effect_rpc/tests/spans.rs @@ -341,12 +341,19 @@ async fn webtransport_call_emits_rpc_request_span() { .unwrap(); let addr = quic.local_addr(); let (_stop, stopped) = tokio::sync::oneshot::channel::<()>(); - tokio::spawn(async move { - let options = Http3Options::default().webtransport(); - serve_h3(&quic, app, limits, options, async { + // Called before the spawn, so the route is registered by the time the client connects. + let served = serve_h3( + &quic, + app, + limits, + Http3Options::default().webtransport(), + async { let _ = stopped.await; - }) - .await + }, + ); + tokio::spawn(async move { + let _quic = quic; + served.await }); let tls = client_tls_config(&identity, None, &[pin]).unwrap(); let request = ConnectRequest { diff --git a/crates/id_effect_rpc/tests/transport_webtransport.rs b/crates/id_effect_rpc/tests/transport_webtransport.rs index 0e61f833..cc5e94ce 100644 --- a/crates/id_effect_rpc/tests/transport_webtransport.rs +++ b/crates/id_effect_rpc/tests/transport_webtransport.rs @@ -153,12 +153,19 @@ fn serve_app(app: axum::Router, config: RpcWebTransportConfig) -> Fixture { .unwrap(); let addr = quic.local_addr(); let (stop, stopped) = tokio::sync::oneshot::channel::<()>(); - tokio::spawn(async move { - let options = Http3Options::default().webtransport(); - serve_h3(&quic, app, limits, options, async { + // Called before the spawn, so the route is registered by the time a test connects. + let served = serve_h3( + &quic, + app, + limits, + Http3Options::default().webtransport(), + async { let _ = stopped.await; - }) - .await + }, + ); + tokio::spawn(async move { + let _quic = quic; + served.await }); Fixture { addr, diff --git a/crates/id_effect_test/tests/containers.rs b/crates/id_effect_test/tests/containers.rs index e2f93ca7..bb5f6374 100644 --- a/crates/id_effect_test/tests/containers.rs +++ b/crates/id_effect_test/tests/containers.rs @@ -237,7 +237,16 @@ fn keycloak_serves_discovery_for_an_imported_realm() { let realm = r#"{"realm": "ledger", "enabled": true}"#; let program = using(|kc: Arc| { let address = kc.container().address(8080).unwrap(); - let discovery = http_get(&address, "/realms/ledger/.well-known/openid-configuration"); + // Docker's port proxy may accept before Keycloak's listener does (an empty reply, seen on + // CI): ask again briefly. + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(20); + let discovery = loop { + let reply = http_get(&address, "/realms/ledger/.well-known/openid-configuration"); + if reply.contains(" 200 ") || std::time::Instant::now() >= deadline { + break reply; + } + std::thread::sleep(std::time::Duration::from_millis(250)); + }; (kc.clone(), discovery) }) .provide_layer( diff --git a/examples/reference/relay/src/node.rs b/examples/reference/relay/src/node.rs index 688bbe79..15eb788f 100644 --- a/examples/reference/relay/src/node.rs +++ b/examples/reference/relay/src/node.rs @@ -308,13 +308,13 @@ impl RelayNode { let addr = endpoint.local_addr(); let (stop_h3, stopped) = oneshot::channel::<()>(); - let served = endpoint.clone(); - let h3_options = h3.clone(); + // Called before the spawn: the h3 route is registered once `start` returns. + let shutdown = async move { + let _ = stopped.await; + }; + let served = serve_h3(&endpoint, app, limits, h3.clone(), shutdown); tokio::spawn(async move { - let shutdown = async move { - let _ = stopped.await; - }; - if let Err(e) = serve_h3(&served, app, limits, h3_options, shutdown).await { + if let Err(e) = served.await { eprintln!("relay: http/3 stopped: {e}"); } });