Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 8 additions & 4 deletions crates/id_effect_axum/src/server/h3.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand All @@ -665,15 +667,17 @@ 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).
pub(crate) fn port(&self) -> u16 {
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<F>(
self,
app: Router,
Expand All @@ -683,8 +687,8 @@ impl HostHttp3 {
where
F: Future<Output = ()> + Send + 'static,
{
let served = serve_h3(
&self.endpoint,
let served = serve_route(
Ok(self.route),
app,
limits,
Http3Options::default(),
Expand Down
10 changes: 6 additions & 4 deletions crates/id_effect_axum/src/server/webtransport.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down
57 changes: 18 additions & 39 deletions crates/id_effect_axum/tests/http3.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -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.
Expand Down
57 changes: 35 additions & 22 deletions crates/id_effect_mail/tests/mailpit.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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::<MailerEnv, _>(message).provide_layer(SmtpMailer::layer::<MailError>(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::<MailError>(config.clone());
match run(deliver::<MailerEnv, _>(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}");
Expand Down
12 changes: 10 additions & 2 deletions crates/id_effect_rpc/tests/browser.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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)
}
Expand Down
17 changes: 12 additions & 5 deletions crates/id_effect_rpc/tests/spans.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down
17 changes: 12 additions & 5 deletions crates/id_effect_rpc/tests/transport_webtransport.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
11 changes: 10 additions & 1 deletion crates/id_effect_test/tests/containers.rs
Original file line number Diff line number Diff line change
Expand Up @@ -237,7 +237,16 @@ fn keycloak_serves_discovery_for_an_imported_realm() {
let realm = r#"{"realm": "ledger", "enabled": true}"#;
let program = using(|kc: Arc<KeycloakContainer>| {
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(
Expand Down
12 changes: 6 additions & 6 deletions examples/reference/relay/src/node.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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}");
}
});
Expand Down
Loading