Implement network lifecycle teardown and logout (#57)
All checks were successful
Native code generation / deterministic (push) Successful in 12m4s
Imaging and meshing gate / native (push) Successful in 4m3s
Native Rust workspace compile / compile (push) Successful in 4m6s

This commit is contained in:
2026-08-09 19:11:21 +00:00
parent 3951bdb809
commit 1fd127bd8a
7 changed files with 1068 additions and 65 deletions

View File

@@ -3,8 +3,10 @@ use libremetaverse::messages::linden::{
EnableSimulatorMessage, EnableSimulatorMessageSimulatorInfoBlock,
};
use libremetaverse::packets::{
DisableSimulatorPacket, Packet, PacketAckPacket, PacketAckPacketPacketsBlock, PacketType,
RegionHandshakePacket, RegionHandshakeReplyPacket, StartPingCheckPacket, UseCircuitCodePacket,
CompleteAgentMovementPacket, DisableSimulatorPacket, LogoutReplyPacket,
LogoutReplyPacketInventoryDataBlock, LogoutRequestPacket, Packet, PacketAckPacket,
PacketAckPacketPacketsBlock, PacketType, RegionHandshakePacket, RegionHandshakeReplyPacket,
StartPingCheckPacket, UseCircuitCodePacket,
};
use libremetaverse::{
CapsEventDictionary, CapsEventQueueCallback, GridClient, Helpers, HttpCapsClient, LoginState,
@@ -20,7 +22,7 @@ use std::collections::{BTreeMap, HashMap};
use std::net::{IpAddr, Ipv4Addr, SocketAddr, UdpSocket};
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::mpsc::{Receiver, Sender, channel};
use std::sync::{Arc, Barrier, Mutex, Weak};
use std::sync::{Arc, Barrier, Mutex, MutexGuard, OnceLock, Weak};
use std::thread::{self, JoinHandle};
use std::time::{Duration, Instant};
@@ -73,6 +75,14 @@ fn wait_until(mut condition: impl FnMut() -> bool) {
}
}
fn network_test_guard() -> MutexGuard<'static, ()> {
static NETWORK_TEST_LOCK: OnceLock<Mutex<()>> = OnceLock::new();
NETWORK_TEST_LOCK
.get_or_init(|| Mutex::new(()))
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
}
#[test]
fn packet_callbacks_preserve_filtering_order_async_policy_and_reentrancy() {
let client = GridClient::new().expect("client");
@@ -355,6 +365,9 @@ enum ServerCommand {
Ping,
Disable,
ExpectEconomy,
Handoff,
ExpectLogout,
LogoutReply(Vec<UUID>),
Stop,
}
@@ -364,6 +377,8 @@ enum ServerReport {
Circuit(u32),
HandshakeReply(u32),
Economy,
Logout(UUID, UUID),
Movement(u32),
Ping(u8),
}
@@ -516,6 +531,93 @@ fn spawn_fake_server() -> FakeServer {
break;
}
},
ServerCommand::Handoff => {
loop {
let (length, _) = socket.recv_from(&mut buffer).expect("UseCircuitCode");
if packet_type(&buffer[..length]) != Some(PacketType::UseCircuitCode) {
continue;
}
let mut position = 0;
let use_circuit = UseCircuitCodePacket::new_with_bytes_int32(
buffer[..length].to_vec(),
&mut position,
)
.unwrap();
report_sender
.send(ServerReport::Circuit(use_circuit.circuit_code.code))
.unwrap();
let sequence = u32::from_be_bytes(buffer[1..5].try_into().unwrap());
socket.send_to(&ack(sequence), client_endpoint).unwrap();
break;
}
loop {
let (length, _) = socket
.recv_from(&mut buffer)
.expect("CompleteAgentMovement");
if packet_type(&buffer[..length]) != Some(PacketType::CompleteAgentMovement)
{
continue;
}
let mut position = 0;
let movement = CompleteAgentMovementPacket::new_with_bytes_int32(
buffer[..length].to_vec(),
&mut position,
)
.unwrap();
report_sender
.send(ServerReport::Movement(movement.agent_data.circuit_code))
.unwrap();
break;
}
}
ServerCommand::ExpectLogout => loop {
let (length, _) = socket.recv_from(&mut buffer).expect("LogoutRequest");
if packet_type(&buffer[..length]) != Some(PacketType::LogoutRequest) {
continue;
}
let mut position = 0;
let request = LogoutRequestPacket::new_with_bytes_int32(
buffer[..length].to_vec(),
&mut position,
)
.unwrap();
report_sender
.send(ServerReport::Logout(
request.agent_data.agent_id,
request.agent_data.session_id,
))
.unwrap();
break;
},
ServerCommand::LogoutReply(inventory_items) => loop {
let (length, _) = socket.recv_from(&mut buffer).expect("LogoutRequest");
if packet_type(&buffer[..length]) != Some(PacketType::LogoutRequest) {
continue;
}
let mut position = 0;
let request = LogoutRequestPacket::new_with_bytes_int32(
buffer[..length].to_vec(),
&mut position,
)
.unwrap();
report_sender
.send(ServerReport::Logout(
request.agent_data.agent_id,
request.agent_data.session_id,
))
.unwrap();
let mut reply = LogoutReplyPacket::new_with_constructor().unwrap();
reply.agent_data.agent_id = request.agent_data.agent_id;
reply.agent_data.session_id = request.agent_data.session_id;
reply.inventory_data = inventory_items
.into_iter()
.map(|item_id| LogoutReplyPacketInventoryDataBlock { item_id })
.collect();
let mut bytes = reply.to_bytes_with_method().unwrap();
bytes[0] &= !(Helpers::MSG_RELIABLE | Helpers::MSG_ZEROCODED);
socket.send_to(&bytes, client_endpoint).unwrap();
break;
},
ServerCommand::Stop => break,
}
}
@@ -592,6 +694,7 @@ fn assert_reuse_preserves_connected_override(
#[test]
fn fake_server_drives_circuit_handshake_ping_disable_and_disconnect_reasons() {
let _network_guard = network_test_guard();
let server = spawn_fake_server();
let client = GridClient::new().expect("client");
let mut manager = NetworkManager::new(client).expect("manager");
@@ -689,6 +792,7 @@ fn fake_server_drives_circuit_handshake_ping_disable_and_disconnect_reasons() {
#[test]
fn concurrent_disconnect_emits_each_transition_once() {
let _network_guard = network_test_guard();
const CALLERS: usize = 12;
let server = spawn_fake_server();
let client = GridClient::new().expect("client");
@@ -741,6 +845,7 @@ fn concurrent_disconnect_emits_each_transition_once() {
#[test]
fn sim_disconnected_callback_can_reenter_disconnect_without_deadlock() {
let _network_guard = network_test_guard();
let server = spawn_fake_server();
let client = GridClient::new().expect("client");
let mut manager = NetworkManager::new(client).expect("manager");
@@ -778,6 +883,7 @@ fn sim_disconnected_callback_can_reenter_disconnect_without_deadlock() {
#[test]
fn keepalive_requires_two_silent_intervals_before_network_timeout() {
let _network_guard = network_test_guard();
let server = spawn_fake_server();
let client = GridClient::new().expect("client");
let mut manager = NetworkManager::new(client).expect("manager");
@@ -821,6 +927,7 @@ fn keepalive_requires_two_silent_intervals_before_network_timeout() {
#[test]
fn enable_simulator_caps_event_adds_each_new_region_once() {
let _network_guard = network_test_guard();
let first_server = spawn_fake_server();
let second_server = spawn_fake_server();
let mut client = GridClient::new().expect("client");
@@ -879,8 +986,446 @@ fn enable_simulator_caps_event_adds_each_new_region_once() {
.unwrap();
}
#[test]
fn connected_region_handoff_preserves_old_sim_and_sends_complete_agent_movement() {
let _network_guard = network_test_guard();
let first_server = spawn_fake_server();
let second_server = spawn_fake_server();
let client = GridClient::new().expect("client");
let mut manager = NetworkManager::new(client).expect("manager");
manager.set_circuit_code(0x7788);
let first = manager
.connect_with_ip_end_point_u_int64_boolean_uri_u_int32_u_int32(
first_server.endpoint,
100,
true,
None,
256,
256,
)
.unwrap()
.unwrap();
let second = manager
.connect_with_ip_end_point_u_int64_boolean_uri_u_int32_u_int32(
second_server.endpoint,
200,
false,
None,
512,
256,
)
.unwrap()
.unwrap();
for server in [&first_server, &second_server] {
assert_eq!(
server.reports.recv_timeout(Duration::from_secs(1)).unwrap(),
ServerReport::Circuit(0x7788)
);
assert_eq!(
server.reports.recv_timeout(Duration::from_secs(1)).unwrap(),
ServerReport::HandshakeReply(0x1 | 0x2 | 0x4)
);
}
let (changed_sender, changed_receiver) = channel();
let _changed = manager.subscribe_sim_changed(Arc::new(move |args| {
changed_sender.send(args.previous_simulator()).unwrap();
}));
second_server.commands.send(ServerCommand::Handoff).unwrap();
let selected = manager
.connect_with_ip_end_point_u_int64_boolean_uri_u_int32_u_int32(
second_server.endpoint,
200,
true,
None,
512,
256,
)
.unwrap()
.unwrap();
assert_eq!(selected, second);
assert_eq!(manager.current_sim(), Some(second.clone()));
assert_eq!(
changed_receiver
.recv_timeout(Duration::from_secs(1))
.unwrap(),
Some(first.clone())
);
assert_eq!(
second_server
.reports
.recv_timeout(Duration::from_secs(1))
.unwrap(),
ServerReport::Circuit(0x7788)
);
assert_eq!(
second_server
.reports
.recv_timeout(Duration::from_secs(1))
.unwrap(),
ServerReport::Movement(0x7788)
);
assert!(first.connected());
assert!(second.connected());
assert_eq!(manager.simulators.len(), 2);
manager
.shutdown_with_disconnect_type(NetworkManagerDisconnectType::ClientInitiated)
.unwrap();
}
#[test]
fn manager_reconnects_with_fresh_workers_after_complete_shutdown() {
let _network_guard = network_test_guard();
let first_server = spawn_fake_server();
let second_server = spawn_fake_server();
let client = GridClient::new().expect("client");
let mut manager = NetworkManager::new(client).expect("manager");
manager.set_circuit_code(0x7878);
manager
.connect_with_ip_end_point_u_int64_boolean_uri_u_int32_u_int32(
first_server.endpoint,
250,
true,
None,
256,
256,
)
.unwrap()
.unwrap();
assert_eq!(
first_server
.reports
.recv_timeout(Duration::from_secs(1))
.unwrap(),
ServerReport::Circuit(0x7878)
);
assert_eq!(
first_server
.reports
.recv_timeout(Duration::from_secs(1))
.unwrap(),
ServerReport::HandshakeReply(0x1 | 0x2 | 0x4)
);
manager
.shutdown_with_disconnect_type(NetworkManagerDisconnectType::ClientInitiated)
.unwrap();
assert!(!manager.connected());
assert!(manager.current_sim().is_none());
assert!(manager.simulators.is_empty());
manager.set_circuit_code(0x7979);
let reconnected = manager
.connect_with_ip_end_point_u_int64_boolean_uri_u_int32_u_int32(
second_server.endpoint,
251,
true,
None,
512,
256,
)
.unwrap()
.unwrap();
assert_eq!(
second_server
.reports
.recv_timeout(Duration::from_secs(1))
.unwrap(),
ServerReport::Circuit(0x7979)
);
assert_eq!(
second_server
.reports
.recv_timeout(Duration::from_secs(1))
.unwrap(),
ServerReport::HandshakeReply(0x1 | 0x2 | 0x4)
);
assert!(manager.connected());
assert_eq!(manager.current_sim(), Some(reconnected));
assert_eq!(manager.simulators.len(), 1);
manager
.shutdown_with_disconnect_type(NetworkManagerDisconnectType::ClientInitiated)
.unwrap();
}
#[test]
fn logout_reply_orders_events_returns_inventory_and_tears_down_once() {
let _network_guard = network_test_guard();
let server = spawn_fake_server();
let client = GridClient::new().expect("client");
let mut manager = NetworkManager::new(client).expect("manager");
manager.set_circuit_code(0x8899);
manager.set_login_seed_capability(Some(Uri("https://caps.invalid/secret".into())));
let login_response = manager
.login_response_data
.as_mut()
.expect("login response storage");
login_response.set_session_id(
UUID::new_with_string("aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa".into()).unwrap(),
);
login_response.set_secure_session_id(
UUID::new_with_string("bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb".into()).unwrap(),
);
login_response.set_mfa_hash("secret-mfa".into());
login_response.set_seed_capability("https://caps.invalid/secret".into());
login_response.set_next_url("https://login.invalid/secret-token".into());
manager
.connect_with_ip_end_point_u_int64_boolean_uri_u_int32_u_int32(
server.endpoint,
300,
true,
None,
256,
256,
)
.unwrap()
.unwrap();
assert_eq!(
server.reports.recv_timeout(Duration::from_secs(1)).unwrap(),
ServerReport::Circuit(0x8899)
);
assert_eq!(
server.reports.recv_timeout(Duration::from_secs(1)).unwrap(),
ServerReport::HandshakeReply(0x1 | 0x2 | 0x4)
);
let order = Arc::new(Mutex::new(Vec::new()));
let logged_order = Arc::clone(&order);
let (inventory_sender, inventory_receiver) = channel();
let _logged_out = manager.subscribe_logged_out(Arc::new(move |args| {
logged_order.lock().unwrap().push("logged_out");
inventory_sender.send(args.inventory_items).unwrap();
}));
let sim_order = Arc::clone(&order);
let _sim_disconnected = manager.subscribe_sim_disconnected(Arc::new(move |_| {
sim_order.lock().unwrap().push("sim_disconnected");
}));
let disconnected_order = Arc::clone(&order);
let disconnected_count = Arc::new(AtomicUsize::new(0));
let count = Arc::clone(&disconnected_count);
let _disconnected = manager.subscribe_disconnected(Arc::new(move |_| {
count.fetch_add(1, Ordering::Relaxed);
disconnected_order.lock().unwrap().push("disconnected");
}));
let inventory = vec![
UUID::new_with_string("11111111-1111-1111-1111-111111111111".into()).unwrap(),
UUID::new_with_string("22222222-2222-2222-2222-222222222222".into()).unwrap(),
];
server
.commands
.send(ServerCommand::LogoutReply(inventory.clone()))
.unwrap();
manager.logout_with_method().unwrap();
assert_eq!(
server.reports.recv_timeout(Duration::from_secs(1)).unwrap(),
ServerReport::Logout(UUID::zero(), UUID::zero())
);
assert_eq!(
inventory_receiver
.recv_timeout(Duration::from_secs(1))
.unwrap(),
inventory
);
assert_eq!(
*order.lock().unwrap(),
["logged_out", "sim_disconnected", "disconnected"]
);
assert!(!manager.connected());
assert!(manager.current_sim().is_none());
assert!(manager.simulators.is_empty());
assert_eq!(manager.circuit_code(), 0);
assert!(manager.login_seed_capability().is_none());
let login_response = manager
.login_response_data
.as_ref()
.expect("login response storage");
assert_eq!(login_response.session_id(), UUID::zero());
assert_eq!(login_response.secure_session_id(), UUID::zero());
assert!(login_response.mfa_hash().is_empty());
assert!(login_response.seed_capability().is_empty());
assert!(login_response.next_url().is_empty());
manager
.shutdown_with_disconnect_type(NetworkManagerDisconnectType::ClientInitiated)
.unwrap();
assert_eq!(disconnected_count.load(Ordering::Relaxed), 1);
}
#[test]
fn async_logout_timeout_and_cancellation_have_distinct_reference_results() {
let _network_guard = network_test_guard();
let timeout_server = spawn_fake_server();
let mut timeout_client = GridClient::new().expect("client");
timeout_client.settings().timing().logout_timeout = 25;
let mut timeout_manager = NetworkManager::new(timeout_client).expect("manager");
timeout_manager.set_circuit_code(0x9911);
timeout_manager
.connect_with_ip_end_point_u_int64_boolean_uri_u_int32_u_int32(
timeout_server.endpoint,
400,
true,
None,
256,
256,
)
.unwrap()
.unwrap();
let _ = timeout_server
.reports
.recv_timeout(Duration::from_secs(1))
.unwrap();
let _ = timeout_server
.reports
.recv_timeout(Duration::from_secs(1))
.unwrap();
let (reason_sender, reason_receiver) = channel();
let _disconnected = timeout_manager.subscribe_disconnected(Arc::new(move |args| {
reason_sender.send(args.reason()).unwrap();
}));
block_on(timeout_manager.logout_with_cancellation_token(None)).unwrap();
assert_eq!(
reason_receiver
.recv_timeout(Duration::from_secs(1))
.unwrap(),
NetworkManagerDisconnectType::NetworkTimeout
);
assert!(!timeout_manager.connected());
let cancel_server = spawn_fake_server();
let mut cancel_client = GridClient::new().expect("client");
cancel_client.settings().timing().logout_timeout = 1_000;
let mut cancel_manager = NetworkManager::new(cancel_client).expect("manager");
cancel_manager.set_circuit_code(0x9922);
cancel_manager
.connect_with_ip_end_point_u_int64_boolean_uri_u_int32_u_int32(
cancel_server.endpoint,
500,
true,
None,
256,
256,
)
.unwrap()
.unwrap();
let _ = cancel_server
.reports
.recv_timeout(Duration::from_secs(1))
.unwrap();
let _ = cancel_server
.reports
.recv_timeout(Duration::from_secs(1))
.unwrap();
let cancellation = CancellationTokenSource::new();
cancellation.cancel();
let result =
block_on(cancel_manager.logout_with_cancellation_token(Some(cancellation.token())));
assert_eq!(result, Err(libremetaverse::Error::Cancelled));
assert!(cancel_manager.connected());
cancel_manager
.shutdown_with_disconnect_type(NetworkManagerDisconnectType::ClientInitiated)
.unwrap();
}
#[test]
fn begin_logout_times_out_nonblocking_and_emits_logged_out_after_teardown() {
let _network_guard = network_test_guard();
let server = spawn_fake_server();
let mut client = GridClient::new().expect("client");
client.settings().timing().logout_timeout = 250;
let mut manager = NetworkManager::new(client).expect("manager");
manager.set_circuit_code(0x9933);
manager
.connect_with_ip_end_point_u_int64_boolean_uri_u_int32_u_int32(
server.endpoint,
600,
true,
None,
256,
256,
)
.unwrap()
.unwrap();
let _ = server.reports.recv_timeout(Duration::from_secs(1)).unwrap();
let _ = server.reports.recv_timeout(Duration::from_secs(1)).unwrap();
let order = Arc::new(Mutex::new(Vec::new()));
let disconnected_order = Arc::clone(&order);
let _disconnected = manager.subscribe_disconnected(Arc::new(move |_| {
disconnected_order.lock().unwrap().push("disconnected");
}));
let logged_order = Arc::clone(&order);
let (logged_sender, logged_receiver) = channel();
let _logged = manager.subscribe_logged_out(Arc::new(move |args| {
logged_order.lock().unwrap().push("logged_out");
logged_sender.send(args.inventory_items).unwrap();
}));
let started = Instant::now();
manager.begin_logout().unwrap();
assert!(started.elapsed() < Duration::from_millis(100));
assert_eq!(
logged_receiver
.recv_timeout(Duration::from_secs(1))
.unwrap(),
Vec::<UUID>::new()
);
assert_eq!(*order.lock().unwrap(), ["disconnected", "logged_out"]);
assert!(!manager.connected());
assert!(manager.simulators.is_empty());
}
#[test]
fn explicit_shutdown_cancels_a_pending_begin_logout_without_retaining_client() {
let _network_guard = network_test_guard();
let server = spawn_fake_server();
let mut client = GridClient::new().expect("client");
client.settings().timing().logout_timeout = 5_000;
let mut manager = NetworkManager::new(client.clone()).expect("manager");
manager.set_circuit_code(0x9944);
manager
.connect_with_ip_end_point_u_int64_boolean_uri_u_int32_u_int32(
server.endpoint,
700,
true,
None,
256,
256,
)
.unwrap()
.unwrap();
let _ = server.reports.recv_timeout(Duration::from_secs(1)).unwrap();
let _ = server.reports.recv_timeout(Duration::from_secs(1)).unwrap();
server.commands.send(ServerCommand::ExpectLogout).unwrap();
manager.begin_logout().unwrap();
assert_eq!(
server.reports.recv_timeout(Duration::from_secs(1)).unwrap(),
ServerReport::Logout(UUID::zero(), UUID::zero())
);
assert_eq!(manager.pending_logout_tasks(), 1);
manager
.shutdown_with_disconnect_type(NetworkManagerDisconnectType::ClientInitiated)
.unwrap();
wait_until(|| manager.pending_logout_tasks() == 0);
assert_eq!(
manager.last_logout_error(),
Some(libremetaverse::Error::Cancelled)
);
assert!(!manager.connected());
assert!(manager.simulators.is_empty());
}
#[test]
fn async_connect_and_shutdown_do_not_require_an_ambient_tokio_runtime() {
let _network_guard = network_test_guard();
let server = spawn_fake_server();
let client = GridClient::new().expect("client");
let mut manager = NetworkManager::new(client).expect("manager");
@@ -911,6 +1456,7 @@ fn async_connect_and_shutdown_do_not_require_an_ambient_tokio_runtime() {
#[test]
fn handshake_timeout_preserves_reference_connect_result_and_shutdown_tears_down_current_sim() {
let _network_guard = network_test_guard();
let (endpoint, server) = spawn_ack_only_server();
let mut client = GridClient::new().expect("client");
client.settings().timing().login_timeout = 25;
@@ -993,6 +1539,7 @@ fn successful_login_response(endpoint: SocketAddr, seed: &str) -> OSD {
#[tokio::test(flavor = "current_thread")]
async fn login_hashes_credentials_parses_response_and_hands_off_session_to_udp_and_caps() {
let _network_guard = network_test_guard();
let server = spawn_fake_server();
let requests = Arc::new(Mutex::new(Vec::<HttpRequest>::new()));
let recorded = Arc::clone(&requests);