Skip to content

Commit 4b1f79c

Browse files
flubflub
authored
fix(tests): bring back MagicEndpoint connect-close test (#1282)
This brings back the MagicEndpoint connect-close test and attempts to make it more reliable. - Most importantly the client sends one message to the server before the server closes the connection. This is a crucial bit of synchronisation for the interaction to be deterministic. - Both client and server run in a task again. Keeps things a bit more local. - The endpoints are now created in the spawned tasks, this ensures we get nice span prefixes on all the logging for each thus distinguishing the log output. - The run_derp_and_stun function is moved into test_utils. Maybe it belongs better there? - The run_derp_and_stun function now returns a drop guard for cleanup. This ensures that whatever happens to the test, regardless of assertions, the cleanup will run. It does mean for a successful test the test will no longer delay exiting before the cleanup is finished (was already not the case for the STUN server). This is fine though, the servers both bind to 127.0.0.1:0 so do not affect anything else while they are still shutting down. - The run_derp_and_stun function now disables IPv6 in the returned DerpMap. Previously it would trigger a DNS lookup. But we only bind on IPv4 so skip this noise. - A couple of minor logging output adjustments. - turn cleanups into "real" drop guards Part of #1183
1 parent 66ec54d commit 4b1f79c

6 files changed

Lines changed: 262 additions & 233 deletions

File tree

‎iroh-net/src/magic_endpoint.rs‎

Lines changed: 169 additions & 151 deletions
Original file line numberDiff line numberDiff line change
@@ -435,155 +435,173 @@ pub async fn get_peer_id(connection: &quinn::Connection) -> anyhow::Result<PeerI
435435
}
436436
}
437437

438-
// TODO: Reenable when not flaky
438+
// TODO: These tests could still be flaky, lets fix that:
439439
// https://github.com/n0-computer/iroh/issues/1183
440-
// #[cfg(tests)]
441-
// mod test {
442-
// use futures::future::BoxFuture;
443-
444-
// use super::{accept_conn, MagicEndpoint};
445-
// use crate::magicsock::conn_tests::{run_derp_and_stun, setup_logging};
446-
447-
// const TEST_ALPN: &[u8] = b"n0/iroh/test";
448-
449-
// async fn setup_pair() -> anyhow::Result<(
450-
// MagicEndpoint,
451-
// MagicEndpoint,
452-
// impl FnOnce() -> BoxFuture<'static, ()>,
453-
// )> {
454-
// let (derp_map, cleanup) = run_derp_and_stun([127, 0, 0, 1].into()).await?;
455-
456-
// let ep1 = MagicEndpoint::builder()
457-
// .alpns(vec![TEST_ALPN.to_vec()])
458-
// .derp_map(Some(derp_map.clone()))
459-
// .bind(0)
460-
// .await?;
461-
462-
// let ep2 = MagicEndpoint::builder()
463-
// .alpns(vec![TEST_ALPN.to_vec()])
464-
// .derp_map(Some(derp_map.clone()))
465-
// .bind(0)
466-
// .await?;
467-
468-
// Ok((ep1, ep2, cleanup))
469-
// }
470-
471-
// #[tokio::test]
472-
// async fn magic_endpoint_connect_close() {
473-
// setup_logging();
474-
// let (ep1, ep2, cleanup) = setup_pair().await.unwrap();
475-
// let peer_id_1 = ep1.peer_id();
476-
477-
// let accept = tokio::spawn(async move {
478-
// let conn = ep1.accept().await.unwrap();
479-
// let (_peer_id, _alpn, conn) = accept_conn(conn).await.unwrap();
480-
// let mut stream = conn.accept_uni().await.unwrap();
481-
// ep1.close(23u8.into(), b"badbadnotgood").await.unwrap();
482-
// let res = conn.accept_uni().await;
483-
// assert_eq!(res.unwrap_err(), quinn::ConnectionError::LocallyClosed);
484-
485-
// let res = stream.read_to_end(10).await;
486-
// assert_eq!(
487-
// res.unwrap_err(),
488-
// quinn::ReadToEndError::Read(quinn::ReadError::ConnectionLost(
489-
// quinn::ConnectionError::LocallyClosed
490-
// ))
491-
// );
492-
// });
493-
494-
// let conn = ep2.connect(peer_id_1, TEST_ALPN, &[]).await.unwrap();
495-
// // open a first stream - this does not error before we accept one stream before closing
496-
// // on the other peer
497-
// let mut stream = conn.open_uni().await.unwrap();
498-
// // now the other peer closed the connection.
499-
// stream.write_all(b"hi").await.unwrap();
500-
// // now the other peer closed the connection.
501-
// let expected_err = quinn::ConnectionError::ApplicationClosed(quinn::ApplicationClose {
502-
// error_code: 23u8.into(),
503-
// reason: b"badbadnotgood".to_vec().into(),
504-
// });
505-
// let err = conn.closed().await;
506-
// assert_eq!(err, expected_err);
507-
508-
// let res = stream.finish().await;
509-
// assert_eq!(
510-
// res.unwrap_err(),
511-
// quinn::WriteError::ConnectionLost(expected_err.clone())
512-
// );
513-
514-
// let res = conn.open_uni().await;
515-
// assert_eq!(res.unwrap_err(), expected_err);
516-
517-
// accept.await.unwrap();
518-
// cleanup().await;
519-
// }
520-
521-
// #[tokio::test]
522-
// async fn magic_endpoint_bidi_send_recv() {
523-
// setup_logging();
524-
// let (ep1, ep2, cleanup) = setup_pair().await.unwrap();
525-
526-
// let peer_id_1 = ep1.peer_id();
527-
// eprintln!("peer id 1 {peer_id_1}");
528-
// let peer_id_2 = ep2.peer_id();
529-
// eprintln!("peer id 2 {peer_id_2}");
530-
531-
// let endpoint = ep2.clone();
532-
// let p2_connect = tokio::spawn(async move {
533-
// let conn = endpoint.connect(peer_id_1, TEST_ALPN, &[]).await.unwrap();
534-
// let (mut send, mut recv) = conn.open_bi().await.unwrap();
535-
// send.write_all(b"hello").await.unwrap();
536-
// send.finish().await.unwrap();
537-
// let m = recv.read_to_end(100).await.unwrap();
538-
// assert_eq!(&m, b"world");
539-
// });
540-
541-
// let endpoint = ep1.clone();
542-
// let p1_accept = tokio::spawn(async move {
543-
// let conn = endpoint.accept().await.unwrap();
544-
// let (peer_id, alpn, conn) = accept_conn(conn).await.unwrap();
545-
// assert_eq!(peer_id, peer_id_2);
546-
// assert_eq!(alpn.as_bytes(), TEST_ALPN);
547-
548-
// let (mut send, mut recv) = conn.accept_bi().await.unwrap();
549-
// let m = recv.read_to_end(100).await.unwrap();
550-
// assert_eq!(m, b"hello");
551-
552-
// send.write_all(b"world").await.unwrap();
553-
// send.finish().await.unwrap();
554-
// });
555-
556-
// let endpoint = ep1.clone();
557-
// let p1_connect = tokio::spawn(async move {
558-
// let conn = endpoint.connect(peer_id_2, TEST_ALPN, &[]).await.unwrap();
559-
// let (mut send, mut recv) = conn.open_bi().await.unwrap();
560-
// send.write_all(b"ola").await.unwrap();
561-
// send.finish().await.unwrap();
562-
// let m = recv.read_to_end(100).await.unwrap();
563-
// assert_eq!(&m, b"mundo");
564-
// });
565-
566-
// let endpoint = ep2.clone();
567-
// let p2_accept = tokio::spawn(async move {
568-
// let conn = endpoint.accept().await.unwrap();
569-
// let (peer_id, alpn, conn) = accept_conn(conn).await.unwrap();
570-
// assert_eq!(peer_id, peer_id_1);
571-
// assert_eq!(alpn.as_bytes(), TEST_ALPN);
572-
573-
// let (mut send, mut recv) = conn.accept_bi().await.unwrap();
574-
// let m = recv.read_to_end(100).await.unwrap();
575-
// assert_eq!(m, b"ola");
576-
577-
// send.write_all(b"mundo").await.unwrap();
578-
// send.finish().await.unwrap();
579-
// });
580-
581-
// p1_accept.await.unwrap();
582-
// p2_connect.await.unwrap();
583-
584-
// p2_accept.await.unwrap();
585-
// p1_connect.await.unwrap();
586-
587-
// cleanup().await;
588-
// }
589-
// }
440+
#[cfg(test)]
441+
mod tests {
442+
use tracing::{info, info_span, Instrument};
443+
444+
use crate::test_utils::{run_derp_and_stun, setup_logging};
445+
446+
use super::*;
447+
448+
const TEST_ALPN: &[u8] = b"n0/iroh/test";
449+
450+
#[tokio::test]
451+
async fn magic_endpoint_connect_close() {
452+
let _guard = setup_logging();
453+
let (derp_map, region_id, _guard) = run_derp_and_stun([127, 0, 0, 1].into()).await.unwrap();
454+
let server_keypair = Keypair::generate();
455+
let server_peer_id = PeerId::from(server_keypair.public());
456+
457+
let server = {
458+
let derp_map = derp_map.clone();
459+
tokio::spawn(
460+
async move {
461+
let ep = MagicEndpoint::builder()
462+
.keypair(server_keypair)
463+
.alpns(vec![TEST_ALPN.to_vec()])
464+
.derp_map(Some(derp_map))
465+
.bind(0)
466+
.await
467+
.unwrap();
468+
info!("accepting connection");
469+
let conn = ep.accept().await.unwrap();
470+
let (_peer_id, _alpn, conn) = accept_conn(conn).await.unwrap();
471+
let mut stream = conn.accept_uni().await.unwrap();
472+
let mut buf = [0u8, 5];
473+
stream.read_exact(&mut buf).await.unwrap();
474+
info!("Accepted 1 stream, received {buf:?}. Closing now.");
475+
ep.close(7u8.into(), b"bye").await.unwrap();
476+
477+
let res = conn.accept_uni().await;
478+
assert_eq!(res.unwrap_err(), quinn::ConnectionError::LocallyClosed);
479+
480+
let res = stream.read_to_end(10).await;
481+
assert_eq!(
482+
res.unwrap_err(),
483+
quinn::ReadToEndError::Read(quinn::ReadError::ConnectionLost(
484+
quinn::ConnectionError::LocallyClosed
485+
))
486+
);
487+
}
488+
.instrument(info_span!("test-server")),
489+
)
490+
};
491+
492+
let client = tokio::spawn(
493+
async move {
494+
let ep = MagicEndpoint::builder()
495+
.alpns(vec![TEST_ALPN.to_vec()])
496+
.derp_map(Some(derp_map))
497+
.bind(0)
498+
.await
499+
.unwrap();
500+
info!("client connecting");
501+
let conn = ep
502+
.connect(server_peer_id, TEST_ALPN, region_id, &[])
503+
.await
504+
.unwrap();
505+
let mut stream = conn.open_uni().await.unwrap();
506+
507+
// First write is accepted by server. We need this bit of synchronisation
508+
// because if the server closes after simply accepting the connection we can
509+
// not be sure our .open_uni() call would succeed as it may already receive
510+
// the error.
511+
stream.write_all(b"hello").await.unwrap();
512+
513+
// Remote now closes the connection, we should see an error sometime soon.
514+
let err = conn.closed().await;
515+
let expected_err =
516+
quinn::ConnectionError::ApplicationClosed(quinn::ApplicationClose {
517+
error_code: 7u8.into(),
518+
reason: b"bye".to_vec().into(),
519+
});
520+
assert_eq!(err, expected_err);
521+
522+
let res = stream.finish().await;
523+
assert_eq!(
524+
res.unwrap_err(),
525+
quinn::WriteError::ConnectionLost(expected_err.clone())
526+
);
527+
528+
let res = conn.open_uni().await;
529+
assert_eq!(res.unwrap_err(), expected_err);
530+
}
531+
.instrument(info_span!("test-client")),
532+
);
533+
534+
let (server, client) = tokio::join!(server, client);
535+
server.unwrap();
536+
client.unwrap();
537+
}
538+
539+
// #[tokio::test]
540+
// async fn magic_endpoint_bidi_send_recv() {
541+
// setup_logging();
542+
// let (ep1, ep2, cleanup) = setup_pair().await.unwrap();
543+
544+
// let peer_id_1 = ep1.peer_id();
545+
// eprintln!("peer id 1 {peer_id_1}");
546+
// let peer_id_2 = ep2.peer_id();
547+
// eprintln!("peer id 2 {peer_id_2}");
548+
549+
// let endpoint = ep2.clone();
550+
// let p2_connect = tokio::spawn(async move {
551+
// let conn = endpoint.connect(peer_id_1, TEST_ALPN, &[]).await.unwrap();
552+
// let (mut send, mut recv) = conn.open_bi().await.unwrap();
553+
// send.write_all(b"hello").await.unwrap();
554+
// send.finish().await.unwrap();
555+
// let m = recv.read_to_end(100).await.unwrap();
556+
// assert_eq!(&m, b"world");
557+
// });
558+
559+
// let endpoint = ep1.clone();
560+
// let p1_accept = tokio::spawn(async move {
561+
// let conn = endpoint.accept().await.unwrap();
562+
// let (peer_id, alpn, conn) = accept_conn(conn).await.unwrap();
563+
// assert_eq!(peer_id, peer_id_2);
564+
// assert_eq!(alpn.as_bytes(), TEST_ALPN);
565+
566+
// let (mut send, mut recv) = conn.accept_bi().await.unwrap();
567+
// let m = recv.read_to_end(100).await.unwrap();
568+
// assert_eq!(m, b"hello");
569+
570+
// send.write_all(b"world").await.unwrap();
571+
// send.finish().await.unwrap();
572+
// });
573+
574+
// let endpoint = ep1.clone();
575+
// let p1_connect = tokio::spawn(async move {
576+
// let conn = endpoint.connect(peer_id_2, TEST_ALPN, &[]).await.unwrap();
577+
// let (mut send, mut recv) = conn.open_bi().await.unwrap();
578+
// send.write_all(b"ola").await.unwrap();
579+
// send.finish().await.unwrap();
580+
// let m = recv.read_to_end(100).await.unwrap();
581+
// assert_eq!(&m, b"mundo");
582+
// });
583+
584+
// let endpoint = ep2.clone();
585+
// let p2_accept = tokio::spawn(async move {
586+
// let conn = endpoint.accept().await.unwrap();
587+
// let (peer_id, alpn, conn) = accept_conn(conn).await.unwrap();
588+
// assert_eq!(peer_id, peer_id_1);
589+
// assert_eq!(alpn.as_bytes(), TEST_ALPN);
590+
591+
// let (mut send, mut recv) = conn.accept_bi().await.unwrap();
592+
// let m = recv.read_to_end(100).await.unwrap();
593+
// assert_eq!(m, b"ola");
594+
595+
// send.write_all(b"mundo").await.unwrap();
596+
// send.finish().await.unwrap();
597+
// });
598+
599+
// p1_accept.await.unwrap();
600+
// p2_connect.await.unwrap();
601+
602+
// p2_accept.await.unwrap();
603+
// p1_connect.await.unwrap();
604+
605+
// cleanup().await;
606+
// }
607+
}

0 commit comments

Comments
 (0)