Lochnair-Labs/irtt-rs
2
0 Fork 0
Code Issues 1 Pull Requests 4 Actions Packages Projects Releases 6 Wiki Activity

refactor(irtt-client): move cadence into managed runner

Browse Source
This commit is contained in:
Nils Andreas Svee committed 2026-10-02 00:03:31 +07:00
1 parent f350c82dae
commit ac29b01fda
31 files changed
+707 -4102
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

crates/irtt-client/src/client/tests/open.rs
+104 -347
View File
@@ -1,37 +1,5 @@
use super::*;
#[test]
fn successful_open_handshake() {
let config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1)));
let params = params_from_config(&config).unwrap();
let server = open_success_server(params.clone());
let mut client = Client::connect(default_test_config(server.addr)).unwrap();
let negotiated = assert_open_started(client.open().unwrap());
assert_eq!(negotiated.params, params);
assert!(client.schedule.is_some());
assert_eq!(client.applied_traffic_class, Some(0));
server.join();
}
#[test]
fn successful_open_restores_configured_receive_timeout() {
let config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1)));
let params = params_from_config(&config).unwrap();
let server = open_success_server(params);
let mut config = default_test_config(server.addr);
config.socket_config.recv_timeout = Some(Duration::from_millis(75));
let mut client = Client::connect(config).unwrap();
assert_open_started(client.open().unwrap());
assert_eq!(
client.test_hooks.last_restored_read_timeout.get(),
Some(Duration::from_millis(75))
);
server.join();
}
#[test]
fn open_fails_when_already_open() {
let config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1)));
@@ -64,84 +32,6 @@ fn open_retries_after_first_timeout() {
server.join();
}
#[test]
fn open_survives_interrupted_receive_within_same_attempt() {
let server = start_fake_server(|socket, tx| {
let (first, peer) = recv_request(&socket, &tx);
let params = Params::decode(&first[4..]).unwrap();
let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, &params, None);
socket.send_to(&reply, peer).unwrap();
});
let config = ClientConfig {
open_timeouts: vec![Duration::from_millis(500)],
..default_test_config(server.addr)
};
let mut client = Client::connect(config).unwrap();
// Several interrupted receive attempts before the real reply arrives
// must not be treated as a timeout, must not retransmit the open
// request, and must not advance to a next configured attempt.
client.test_hooks.inject_recv_interrupted.set(5);
let outcome = client.open().unwrap();
assert_open_started(outcome);
// Exactly one request reached the server: interruptions were absorbed
// inside the single configured attempt.
assert_eq!(server.rx.iter().take(1).count(), 1);
server.join();
}
#[test]
fn open_interrupted_receive_does_not_extend_the_attempt_deadline() {
// A fake server that never replies: with a short attempt timeout, the
// real question is whether interruptions can push the effective wait
// past the configured deadline. They must not: the deadline stays
// absolute across any number of interruptions.
let server = start_fake_server(|socket, tx| {
let _ = recv_request(&socket, &tx);
});
let config = ClientConfig {
open_timeouts: vec![Duration::from_millis(200)],
..default_test_config(server.addr)
};
let mut client = Client::connect(config).unwrap();
// Each interruption carries no timing information of its own, so the
// loop must keep re-checking the real deadline rather than treating an
// interruption as "more time available" or a reason to fail outright.
client.test_hooks.inject_recv_interrupted.set(20);
let started = Instant::now();
let result = client.open();
let elapsed = started.elapsed();
assert!(matches!(result, Err(ClientError::OpenTimeout)));
assert!(
elapsed < Duration::from_millis(800),
"interrupted receives must not extend the attempt deadline, took {elapsed:?}"
);
server.join();
}
#[test]
fn open_timeout_after_all_timeouts() {
let server = timeout_server(Duration::from_millis(700));
let config = ClientConfig {
open_timeouts: vec![Duration::from_millis(200), Duration::from_millis(200)],
socket_config: crate::SocketConfig {
recv_timeout: Some(Duration::from_millis(50)),
..Default::default()
},
..default_test_config(server.addr)
};
let mut client = Client::connect(config).unwrap();
assert!(matches!(client.open(), Err(ClientError::OpenTimeout)));
assert_eq!(server.rx.iter().take(2).count(), 2);
assert_eq!(
client.test_hooks.last_restored_read_timeout.get(),
Some(Duration::from_millis(50))
);
server.join();
}
#[test]
fn protocol_version_mismatch_fails() {
let mut config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1)));
@@ -175,41 +65,6 @@ fn server_rejection_fails_in_normal_mode() {
server.join();
}
#[test]
fn multiple_untrusted_datagrams_before_valid_reply_use_one_attempt() {
let params = default_params();
let server = start_fake_server(move |socket, tx| {
let (_, peer) = recv_request(&socket, &tx);
let valid = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, &params, None);
let mut wrong_type = valid.clone();
wrong_type[3] = FLAG_REPLY;
let mut reserved_flags = valid.clone();
reserved_flags[3] |= 0x10;
let mut bad_magic = valid.clone();
bad_magic[0] ^= 0xff;
let unexpected_hmac =
open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, &params, Some(b"unexpected"));
for packet in [
vec![0_u8],
bad_magic,
wrong_type,
reserved_flags,
unexpected_hmac,
valid,
] {
socket.send_to(&packet, peer).unwrap();
}
});
let mut client = Client::connect(default_test_config(server.addr)).unwrap();
assert_open_started(client.open().unwrap());
assert_eq!(server.rx.iter().take(1).count(), 1);
assert!(server.rx.try_recv().is_err());
assert!(client.schedule.is_some());
server.join();
}
#[test]
fn ignored_datagrams_do_not_restart_the_attempt_deadline() {
let server = start_fake_server(move |socket, tx| {
@@ -237,60 +92,6 @@ fn ignored_datagrams_do_not_restart_the_attempt_deadline() {
server.join();
}
#[test]
fn only_ignored_open_datagrams_eventually_time_out() {
let server = start_fake_server(move |socket, tx| {
for _ in 0..2 {
let (_, peer) = recv_request(&socket, &tx);
socket.send_to(&[0_u8], peer).unwrap();
socket.send_to(&MAGIC, peer).unwrap();
}
thread::sleep(Duration::from_millis(300));
});
let config = ClientConfig {
open_timeouts: vec![Duration::from_millis(200), Duration::from_millis(200)],
socket_config: crate::SocketConfig {
recv_timeout: Some(Duration::from_millis(50)),
..Default::default()
},
..default_test_config(server.addr)
};
let mut client = Client::connect(config).unwrap();
assert!(matches!(client.open(), Err(ClientError::OpenTimeout)));
assert_eq!(server.rx.iter().take(2).count(), 2);
assert!(server.rx.try_recv().is_err());
assert_eq!(
client.test_hooks.last_restored_read_timeout.get(),
Some(Duration::from_millis(50))
);
server.join();
}
#[test]
fn trusted_zero_token_normal_reply_is_terminal() {
let params = default_params();
let server = start_fake_server(move |socket, tx| {
let (_, peer) = recv_request(&socket, &tx);
// encode_open_reply rejects a zero token without FLAG_CLOSE, so this
// deliberately non-compliant reply is built by encoding a normal
// reply with a placeholder token and then zeroing the token field.
let mut reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, &params, None);
reply[HMAC_OFFSET..HMAC_OFFSET + 8].copy_from_slice(&0_u64.to_le_bytes());
socket.send_to(&reply, peer).unwrap();
});
let mut client = Client::connect(default_test_config(server.addr)).unwrap();
assert!(matches!(
client.open(),
Err(ClientError::Protocol(irtt_proto::ProtoError::ZeroToken))
));
assert_eq!(server.rx.iter().take(1).count(), 1);
assert!(client.runtime.prepare_open_request().is_ok());
assert!(client.schedule.is_none());
server.join();
}
#[test]
fn opening_deadline_overflow_occurs_before_send() {
let server = timeout_server(Duration::from_millis(250));
@@ -305,47 +106,6 @@ fn opening_deadline_overflow_occurs_before_send() {
server.join();
}
#[test]
fn schedule_duration_overflow_cannot_partially_open() {
let server = start_fake_server(|_socket, _tx| {});
let client = Client::connect(default_test_config(server.addr)).unwrap();
let reply = irtt_proto::OpenReply {
flags: FLAG_OPEN | FLAG_REPLY,
token: TOKEN,
params: params_from_config(client.runtime.config()).unwrap(),
};
let mut latest = Instant::now();
let mut step = Duration::MAX;
while !step.is_zero() {
if let Some(candidate) = latest.checked_add(step) {
latest = candidate;
}
step /= 2;
}
let opened_at = ClientTimestamp {
mono: latest,
wall: SystemTime::now(),
};
let machine = client
.runtime
.prepare_open_acceptance(reply, opened_at)
.unwrap();
let previous_recv_len = client.recv_buffer.len();
let failure = client.prepare_client_open(machine, opened_at).unwrap_err();
assert!(matches!(
failure.primary,
ClientError::NegotiationRejected { .. }
));
assert!(failure.machine.cleanup_close_packet().is_some());
assert!(client.runtime.prepare_open_request().is_ok());
assert!(client.schedule.is_none());
assert_eq!(client.applied_traffic_class, None);
assert_eq!(client.recv_buffer.len(), previous_recv_len);
server.join();
}
#[test]
fn post_token_negotiation_failure_sends_cleanup_close() {
let mut returned = default_params();
@@ -377,146 +137,143 @@ fn post_token_negotiation_failure_sends_cleanup_close() {
}
#[test]
fn cleanup_send_failure_does_not_replace_primary_open_error() {
let mut returned = default_params();
returned.interval_ns += 1;
let server = start_fake_server(move |socket, tx| {
let (_, peer) = recv_request(&socket, &tx);
socket
.send_to(
&open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, &returned, None),
peer,
)
.unwrap();
});
fn successful_open_handshake() {
let config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1)));
let params = params_from_config(&config).unwrap();
let server = open_success_server(params.clone());
let mut client = Client::connect(default_test_config(server.addr)).unwrap();
client.test_hooks.fail_cleanup_send.set(true);
assert!(matches!(
client.open(),
Err(ClientError::NegotiationRejected { .. })
));
assert!(client.runtime.prepare_open_request().is_ok());
let negotiated = assert_open_started(client.open().unwrap());
assert_eq!(negotiated.params, params);
assert_eq!(client.applied_traffic_class, Some(0));
server.join();
}
#[test]
fn receive_buffer_reservation_failure_precedes_dscp_application() {
let mut returned = default_params();
// Raw wire TOS/Traffic Class byte for DSCP codepoint 46 (EF), matching
// what the client below requests so negotiation succeeds unchanged.
returned.dscp = 184;
let server = start_fake_server(move |socket, tx| {
let (_, peer) = recv_request(&socket, &tx);
socket
.send_to(
&open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, &returned, None),
peer,
)
.unwrap();
let _ = recv_request(&socket, &tx);
});
fn successful_open_restores_configured_receive_timeout() {
let config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1)));
let params = params_from_config(&config).unwrap();
let server = open_success_server(params);
let mut config = default_test_config(server.addr);
config.dscp = 46;
config.socket_config.recv_timeout = Some(Duration::from_millis(75));
let mut client = Client::connect(config).unwrap();
client
.test_hooks
.recv_buffer_len_override
.set(Some(usize::MAX));
client.test_hooks.fail_open_dscp.set(true);
assert!(matches!(
client.open(),
Err(ClientError::AllocationFailed {
operation: "negotiated receive buffer",
..
})
));
assert!(client.test_hooks.fail_open_dscp.get());
assert!(client.runtime.prepare_open_request().is_ok());
assert!(client.schedule.is_none());
assert_eq!(client.applied_traffic_class, None);
assert_open_started(client.open().unwrap());
assert_eq!(
client.socket.read_timeout().unwrap(),
Some(Duration::from_millis(75))
);
server.join();
}
#[test]
fn dscp_application_failure_leaves_machine_connected() {
let params = default_params();
let server = start_fake_server(move |socket, tx| {
let (_, peer) = recv_request(&socket, &tx);
socket
.send_to(
&open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, &params, None),
peer,
)
.unwrap();
let _ = recv_request(&socket, &tx);
});
let mut client = Client::connect(default_test_config(server.addr)).unwrap();
client.test_hooks.fail_open_dscp.set(true);
assert!(matches!(
client.open(),
Err(ClientError::SocketOption {
operation: "set negotiated DSCP",
..
})
));
assert!(client.test_hooks.prepared_active_session_before_dscp.get());
assert!(client.runtime.prepare_open_request().is_ok());
assert!(client.schedule.is_none());
assert_eq!(client.applied_traffic_class, None);
fn open_timeout_after_all_timeouts() {
let server = timeout_server(Duration::from_millis(700));
let config = ClientConfig {
open_timeouts: vec![Duration::from_millis(200), Duration::from_millis(200)],
socket_config: crate::SocketConfig {
recv_timeout: Some(Duration::from_millis(50)),
..Default::default()
},
..default_test_config(server.addr)
};
let mut client = Client::connect(config).unwrap();
assert!(matches!(client.open(), Err(ClientError::OpenTimeout)));
assert_eq!(server.rx.iter().take(2).count(), 2);
assert_eq!(
client.socket.read_timeout().unwrap(),
Some(Duration::from_millis(50))
);
server.join();
}
#[test]
fn read_timeout_restoration_failure_leaves_machine_connected() {
fn multiple_untrusted_datagrams_before_valid_reply_use_one_attempt() {
let params = default_params();
let server = start_fake_server(move |socket, tx| {
let (_, peer) = recv_request(&socket, &tx);
socket
.send_to(
&open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, &params, None),
peer,
)
.unwrap();
let _ = recv_request(&socket, &tx);
let valid = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, &params, None);
let mut wrong_type = valid.clone();
wrong_type[3] = FLAG_REPLY;
let mut reserved_flags = valid.clone();
reserved_flags[3] |= 0x10;
let mut bad_magic = valid.clone();
bad_magic[0] ^= 0xff;
let unexpected_hmac =
open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, &params, Some(b"unexpected"));
for packet in [
vec![0_u8],
bad_magic,
wrong_type,
reserved_flags,
unexpected_hmac,
valid,
] {
socket.send_to(&packet, peer).unwrap();
}
});
let mut client = Client::connect(default_test_config(server.addr)).unwrap();
client.test_hooks.fail_open_timeout_restore.set(true);
assert!(matches!(
client.open(),
Err(ClientError::ReadTimeoutRestore { .. })
));
assert!(client.runtime.prepare_open_request().is_ok());
assert!(client.schedule.is_none());
assert_eq!(client.applied_traffic_class, None);
assert_open_started(client.open().unwrap());
assert_eq!(server.rx.iter().take(1).count(), 1);
assert!(server.rx.try_recv().is_err());
server.join();
}
#[test]
fn dscp_rollback_failure_keeps_timeout_restoration_primary() {
fn only_ignored_open_datagrams_eventually_time_out() {
let server = start_fake_server(move |socket, tx| {
for _ in 0..2 {
let (_, peer) = recv_request(&socket, &tx);
socket.send_to(&[0_u8], peer).unwrap();
socket.send_to(&MAGIC, peer).unwrap();
}
thread::sleep(Duration::from_millis(300));
});
let config = ClientConfig {
open_timeouts: vec![Duration::from_millis(200), Duration::from_millis(200)],
socket_config: crate::SocketConfig {
recv_timeout: Some(Duration::from_millis(50)),
..Default::default()
},
..default_test_config(server.addr)
};
let mut client = Client::connect(config).unwrap();
assert!(matches!(client.open(), Err(ClientError::OpenTimeout)));
assert_eq!(server.rx.iter().take(2).count(), 2);
assert!(server.rx.try_recv().is_err());
assert_eq!(
client.socket.read_timeout().unwrap(),
Some(Duration::from_millis(50))
);
server.join();
}
#[test]
fn trusted_zero_token_normal_reply_is_terminal() {
let params = default_params();
let server = start_fake_server(move |socket, tx| {
let (_, peer) = recv_request(&socket, &tx);
socket
.send_to(
&open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, &params, None),
peer,
)
.unwrap();
let _ = recv_request(&socket, &tx);
// encode_open_reply rejects a zero token without FLAG_CLOSE, so this
// deliberately non-compliant reply is built by encoding a normal
// reply with a placeholder token and then zeroing the token field.
let mut reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, &params, None);
reply[HMAC_OFFSET..HMAC_OFFSET + 8].copy_from_slice(&0_u64.to_le_bytes());
socket.send_to(&reply, peer).unwrap();
});
let mut client = Client::connect(default_test_config(server.addr)).unwrap();
client.test_hooks.fail_open_timeout_restore.set(true);
client.test_hooks.fail_dscp_restore.set(true);
let error = client.open().unwrap_err();
assert!(matches!(error, ClientError::ReadTimeoutRestore { .. }));
assert!(matches!(
client.open(),
Err(ClientError::Protocol(irtt_proto::ProtoError::ZeroToken))
));
assert_eq!(server.rx.iter().take(1).count(), 1);
assert!(client.runtime.prepare_open_request().is_ok());
assert!(client.schedule.is_none());
assert_eq!(client.applied_traffic_class, None);
server.join();
}