test: keep one historical KCP compatibility baseline
CI / Lint (push) Canceled after 0s
CI / Linux arm64 (debug) (push) Canceled after 0s
CI / Linux arm64 (release) (push) Canceled after 0s
CI / Linux x64 (debug) (push) Canceled after 0s
CI / Linux x64 (release) (push) Canceled after 0s
CI / Windows arm64 (debug) (push) Canceled after 0s
CI / Windows arm64 (release) (push) Canceled after 0s
CI / Windows x64 (debug) (push) Canceled after 0s
CI / Windows x64 (release) (push) Canceled after 0s
CI / macOS arm64 (debug) (push) Canceled after 0s
CI / macOS arm64 (release) (push) Canceled after 0s
CI / macOS x64 (debug) (push) Canceled after 0s
CI / macOS x64 (release) (push) Canceled after 0s
CI / Cross aarch64-unknown-linux-musl (debug) (push) Canceled after 0s
CI / Cross riscv64gc-unknown-linux-musl (debug) (push) Canceled after 0s
CI / Cross x86_64-unknown-linux-musl (debug) (push) Canceled after 0s
CI / Cross aarch64-unknown-linux-musl (release) (push) Canceled after 0s
CI / Cross riscv64gc-unknown-linux-musl (release) (push) Canceled after 0s
CI / Cross x86_64-unknown-linux-musl (release) (push) Canceled after 0s

Remove the intermediate b37ee660 dev-dependency and its duplicate
compatibility cases. Keep the pre-negotiation d7427c22 endpoint as
the maintained legacy baseline.

Flatten the single-use test macro into ordinary test functions and
retain negotiation, bidirectional data, half-close, and duplicate SYN
assertions. Protocol implementation files are unchanged.

Validation: all 31 tests and strict Clippy pass.
This commit is contained in:
KKRainbow
2026-09-14 22:04:54 +08:00
parent 3ef5c4161f
commit 268533568d
2 changed files with 165 additions and 186 deletions
+1 -1
View File
@@ -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" }
+164 -185
View File
@@ -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(&current_id).unwrap(), b"metadata"[..]);
assert_eq!(legacy.conn_data(&legacy_id).unwrap(), b"metadata"[..]);
let mut current_stream =
kcp_sys::stream::KcpStream::new(&current, 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(&current_id).unwrap(), b"metadata"[..]);
assert_eq!(legacy.conn_data(&legacy_id).unwrap(), b"metadata"[..]);
let mut current_stream = kcp_sys::stream::KcpStream::new(&current, 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;
}