diff --git a/Cargo.toml b/Cargo.toml index 5b6b27f..f278356 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -32,5 +32,5 @@ cc = "1.2.10" [dev-dependencies] tokio = { version = "1", features = ["test-util"] } +# Run compatibility tests against the pre-negotiation implementation. kcp-sys-legacy = { package = "kcp-sys", git = "https://github.com/EasyTier/kcp-sys.git", rev = "d7427c22d764deb1860a7d37acc446ed5033464c" } -kcp-sys-baseline = { package = "kcp-sys", git = "https://github.com/EasyTier/kcp-sys.git", rev = "b37ee660fb70bb6d816fb8bbc08b140e55e7218b" } diff --git a/tests/legacy_compat.rs b/tests/legacy_compat.rs index 69ef906..bffc9a9 100644 --- a/tests/legacy_compat.rs +++ b/tests/legacy_compat.rs @@ -74,194 +74,173 @@ async fn write_chunks(stream: &mut (impl AsyncWrite + Unpin), bytes: &[u8]) { } } -// Both dependencies are real published implementations; the macro shares only -// the wire bridge and assertions, never substitutes a legacy state machine. -macro_rules! compatibility_tests { - ($module:ident, $dependency:ident, $broken_pipe_on_eof:expr) => { - mod $module { - use super::*; - use baseline::endpoint::KcpEndpoint; - use $dependency as baseline; +async fn mixed_pair(new_source: bool, duplicate_syn: bool) { + let mut current = kcp_sys::endpoint::KcpEndpoint::new(); + let mut legacy = KcpEndpoint::new(); + current.run().await; + legacy.run().await; + // Let the old endpoint's immediate first heartbeat tick run with no state. + // Its known PING-before-SYN startup race is unrelated to mode negotiation. + tokio::time::sleep(Duration::from_millis(1)).await; - async fn mixed_pair(new_source: bool, duplicate_syn: bool) { - let mut current = kcp_sys::endpoint::KcpEndpoint::new(); - let mut legacy = KcpEndpoint::new(); - current.run().await; - legacy.run().await; - // Let the old endpoint's immediate first heartbeat tick run with no state. - // Its known PING-before-SYN startup race is unrelated to mode negotiation. - tokio::time::sleep(Duration::from_millis(1)).await; - - let captured = Arc::new(Mutex::new(Vec::new())); - let mut tasks = JoinSet::new(); - let mut current_output = current.output_receiver().unwrap(); - let legacy_input = legacy.input_sender(); - let packets = captured.clone(); - tasks.spawn(async move { - while let Some(packet) = current_output.recv().await { - let bytes = packet.inner(); - let copies = if duplicate_syn && bytes[FLAGS_OFFSET] == SYN { - 2 - } else { - 1 - }; - for _ in 0..copies { - packets.lock().push((FromEndpoint::Current, bytes.clone())); - if legacy_input.send(bytes.clone().into()).await.is_err() { - return; - } - } - } - }); - let mut legacy_output = legacy.output_receiver().unwrap(); - let current_input = current.input_sender(); - let packets = captured.clone(); - tasks.spawn(async move { - while let Some(packet) = legacy_output.recv().await { - let bytes = packet.inner(); - let copies = if duplicate_syn && bytes[FLAGS_OFFSET] == SYN { - 2 - } else { - 1 - }; - for _ in 0..copies { - packets.lock().push((FromEndpoint::Legacy, bytes.clone())); - if current_input.send(bytes.clone().into()).await.is_err() { - return; - } - } - } - }); - - let (current_id, legacy_id) = timeout(Duration::from_secs(5), async { - if new_source { - tokio::try_join!( - async { - current - .connect( - Duration::from_secs(1), - 1, - 2, - Bytes::from_static(b"metadata"), - ) - .await - .map_err(anyhow::Error::from) - }, - async { legacy.accept().await.map_err(anyhow::Error::from) }, - ) - } else { - tokio::try_join!( - async { current.accept().await.map_err(anyhow::Error::from) }, - async { - legacy - .connect( - Duration::from_secs(1), - 1, - 2, - Bytes::from_static(b"metadata"), - ) - .await - .map_err(anyhow::Error::from) - }, - ) - } - }) - .await - .expect("mixed-version handshake stalled") - .expect("mixed-version handshake failed"); - assert_eq!(current.conn_data(¤t_id).unwrap(), b"metadata"[..]); - assert_eq!(legacy.conn_data(&legacy_id).unwrap(), b"metadata"[..]); - let mut current_stream = - kcp_sys::stream::KcpStream::new(¤t, current_id).unwrap(); - let mut legacy_stream = - baseline::stream::KcpStream::new(&legacy, legacy_id).unwrap(); - - timeout(Duration::from_secs(5), async { - let request = vec![0x35; 256 * 1024]; - let mut received = vec![0; request.len()]; - let ((), read) = tokio::join!( - write_chunks(&mut current_stream, &request), - legacy_stream.read_exact(&mut received), - ); - read.unwrap(); - assert_eq!(received, request); - - current_stream.shutdown().await.unwrap(); - if $broken_pipe_on_eof { - // d7427c2 reports BrokenPipe for orderly receive closure. - // Preserve that known limitation instead of asserting a new EOF API. - let error = legacy_stream.read_u8().await.unwrap_err(); - assert_eq!(error.kind(), std::io::ErrorKind::BrokenPipe); - } else { - assert_eq!(legacy_stream.read(&mut [0]).await.unwrap(), 0); - } - - let response = vec![0xa7; 256 * 1024]; - let mut received = vec![0; response.len()]; - let ((), read) = tokio::join!( - write_chunks(&mut legacy_stream, &response), - current_stream.read_exact(&mut received), - ); - read.unwrap(); - assert_eq!(received, response); - legacy_stream.shutdown().await.unwrap(); - assert_eq!(current_stream.read(&mut [0]).await.unwrap(), 0); - }) - .await - .expect("mixed-version data or half-close stalled"); - - if duplicate_syn { - if new_source { - assert!(timeout(Duration::from_millis(50), legacy.accept()) - .await - .is_err()); - } else { - assert!(timeout(Duration::from_millis(50), current.accept()) - .await - .is_err()); - } + let captured = Arc::new(Mutex::new(Vec::new())); + let mut tasks = JoinSet::new(); + let mut current_output = current.output_receiver().unwrap(); + let legacy_input = legacy.input_sender(); + let packets = captured.clone(); + tasks.spawn(async move { + while let Some(packet) = current_output.recv().await { + let bytes = packet.inner(); + let copies = if duplicate_syn && bytes[FLAGS_OFFSET] == SYN { + 2 + } else { + 1 + }; + for _ in 0..copies { + packets.lock().push((FromEndpoint::Current, bytes.clone())); + if legacy_input.send(bytes.clone().into()).await.is_err() { + return; } - - let captured = captured.lock(); - assert!(captured - .iter() - .any(|(_, packet)| packet[FLAGS_OFFSET] == FIN)); - assert!(!captured - .iter() - .any(|(_, packet)| packet[FLAGS_OFFSET] & FIN_ACK != 0)); - assert!(!captured - .iter() - .any(|(_, packet)| packet[FLAGS_OFFSET] & RST != 0)); - for (from, packet) in captured.iter() { - let offered_mode = - new_source && *from == FromEndpoint::Current && packet[FLAGS_OFFSET] == SYN; - assert_eq!( - packet[MODE_OFFSET], - u8::from(offered_mode), - "unexpected mode from {from:?}, flags={:#x}", - packet[FLAGS_OFFSET] - ); - } - } - - #[tokio::test] - async fn new_source_negotiates_legacy_with_real_old_destination() { - mixed_pair(true, false).await; - } - - #[tokio::test] - async fn real_old_source_connects_to_new_destination_without_mode_offer() { - mixed_pair(false, false).await; - } - - #[tokio::test] - async fn duplicate_syn_in_mixed_connections_does_not_reset_or_accept_twice() { - mixed_pair(true, true).await; - mixed_pair(false, true).await; } } - }; + }); + let mut legacy_output = legacy.output_receiver().unwrap(); + let current_input = current.input_sender(); + let packets = captured.clone(); + tasks.spawn(async move { + while let Some(packet) = legacy_output.recv().await { + let bytes = packet.inner(); + let copies = if duplicate_syn && bytes[FLAGS_OFFSET] == SYN { + 2 + } else { + 1 + }; + for _ in 0..copies { + packets.lock().push((FromEndpoint::Legacy, bytes.clone())); + if current_input.send(bytes.clone().into()).await.is_err() { + return; + } + } + } + }); + + let (current_id, legacy_id) = timeout(Duration::from_secs(5), async { + if new_source { + tokio::try_join!( + async { + current + .connect( + Duration::from_secs(1), + 1, + 2, + Bytes::from_static(b"metadata"), + ) + .await + .map_err(anyhow::Error::from) + }, + async { legacy.accept().await.map_err(anyhow::Error::from) }, + ) + } else { + tokio::try_join!( + async { current.accept().await.map_err(anyhow::Error::from) }, + async { + legacy + .connect( + Duration::from_secs(1), + 1, + 2, + Bytes::from_static(b"metadata"), + ) + .await + .map_err(anyhow::Error::from) + }, + ) + } + }) + .await + .expect("mixed-version handshake stalled") + .expect("mixed-version handshake failed"); + assert_eq!(current.conn_data(¤t_id).unwrap(), b"metadata"[..]); + assert_eq!(legacy.conn_data(&legacy_id).unwrap(), b"metadata"[..]); + let mut current_stream = kcp_sys::stream::KcpStream::new(¤t, current_id).unwrap(); + let mut legacy_stream = kcp_sys_legacy::stream::KcpStream::new(&legacy, legacy_id).unwrap(); + + timeout(Duration::from_secs(5), async { + let request = vec![0x35; 256 * 1024]; + let mut received = vec![0; request.len()]; + let ((), read) = tokio::join!( + write_chunks(&mut current_stream, &request), + legacy_stream.read_exact(&mut received), + ); + read.unwrap(); + assert_eq!(received, request); + + current_stream.shutdown().await.unwrap(); + // d7427c2 reports BrokenPipe for orderly receive closure. + // Preserve that known limitation instead of asserting a new EOF API. + let error = legacy_stream.read_u8().await.unwrap_err(); + assert_eq!(error.kind(), std::io::ErrorKind::BrokenPipe); + + let response = vec![0xa7; 256 * 1024]; + let mut received = vec![0; response.len()]; + let ((), read) = tokio::join!( + write_chunks(&mut legacy_stream, &response), + current_stream.read_exact(&mut received), + ); + read.unwrap(); + assert_eq!(received, response); + legacy_stream.shutdown().await.unwrap(); + assert_eq!(current_stream.read(&mut [0]).await.unwrap(), 0); + }) + .await + .expect("mixed-version data or half-close stalled"); + + if duplicate_syn { + if new_source { + assert!(timeout(Duration::from_millis(50), legacy.accept()) + .await + .is_err()); + } else { + assert!(timeout(Duration::from_millis(50), current.accept()) + .await + .is_err()); + } + } + + let captured = captured.lock(); + assert!(captured + .iter() + .any(|(_, packet)| packet[FLAGS_OFFSET] == FIN)); + assert!(!captured + .iter() + .any(|(_, packet)| packet[FLAGS_OFFSET] & FIN_ACK != 0)); + assert!(!captured + .iter() + .any(|(_, packet)| packet[FLAGS_OFFSET] & RST != 0)); + for (from, packet) in captured.iter() { + let offered_mode = + new_source && *from == FromEndpoint::Current && packet[FLAGS_OFFSET] == SYN; + assert_eq!( + packet[MODE_OFFSET], + u8::from(offered_mode), + "unexpected mode from {from:?}, flags={:#x}", + packet[FLAGS_OFFSET] + ); + } } -compatibility_tests!(original_legacy, kcp_sys_legacy, true); -compatibility_tests!(half_close_baseline, kcp_sys_baseline, false); +#[tokio::test] +async fn new_source_negotiates_legacy_with_real_old_destination() { + mixed_pair(true, false).await; +} + +#[tokio::test] +async fn real_old_source_connects_to_new_destination_without_mode_offer() { + mixed_pair(false, false).await; +} + +#[tokio::test] +async fn duplicate_syn_in_mixed_connections_does_not_reset_or_accept_twice() { + mixed_pair(true, true).await; + mixed_pair(false, true).await; +}