diff --git a/soulseek-rs-lib/src/client/rooms.rs b/soulseek-rs-lib/src/client/rooms.rs index d11da0c..671ff53 100644 --- a/soulseek-rs-lib/src/client/rooms.rs +++ b/soulseek-rs-lib/src/client/rooms.rs @@ -150,8 +150,11 @@ impl Client { /// Returns [`SoulseekRs::NotConnected`] if the client is not connected. pub fn watch_user(&self, username: &str) -> Result<()> { use crate::message::server::MessageFactory; + // Held across the send: the reply is applied under this lock, so it + // cannot land first and have its "does not exist" undone by the add. + let mut context = self.context.write_safe()?; self.send_server_message(MessageFactory::build_watch_user(username))?; - self.context.write_safe()?.add_watched_user(username); + context.add_watched_user(username); Ok(()) } diff --git a/soulseek-rs-lib/tests/e2e.rs b/soulseek-rs-lib/tests/e2e.rs index 5727b75..d859fa5 100644 --- a/soulseek-rs-lib/tests/e2e.rs +++ b/soulseek-rs-lib/tests/e2e.rs @@ -76,38 +76,51 @@ impl TestServer { } /// Spawn a local soulfind on an ephemeral port with a throwaway database. + /// + /// soulfind exits if its port is taken between `free_port()` handing it + /// out and soulfind binding it, so a child that dies is retried on a + /// fresh port rather than read as a missing server. fn spawn(gate: std::sync::MutexGuard<'static, ()>) -> Option { let bin = soulfind_binary()?; - let port = free_port()?; - let db = std::env::temp_dir().join(format!("soulfind-e2e-{port}.db")); - let _ = std::fs::remove_file(&db); + for _ in 0..3 { + let port = free_port()?; + let db = + std::env::temp_dir().join(format!("soulfind-e2e-{port}.db")); + let _ = std::fs::remove_file(&db); - let mut child = Command::new(&bin) - .arg("-p") - .arg(port.to_string()) - .arg("-d") - .arg(&db) - .stdout(Stdio::null()) - .stderr(Stdio::null()) - .spawn() - .ok()?; + let mut child = Command::new(&bin) + .arg("-p") + .arg(port.to_string()) + .arg("-d") + .arg(&db) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .ok()?; - if wait_until_listening("127.0.0.1", port, Duration::from_secs(5)) - .is_none() - { - // Server never came up (e.g. a toolchain/SQLite issue); skip. + // A child that has exited is never taken as up: soulfind exits + // when its port is gone, and whoever took it may answer instead. + let deadline = Instant::now() + Duration::from_secs(10); + while Instant::now() < deadline + && child.try_wait().ok().flatten().is_none() + { + if TcpStream::connect(("127.0.0.1", port)).is_ok() { + return Some(Self { + host: "127.0.0.1".to_string(), + port, + child: Some(child), + db: Some(db), + _gate: gate, + }); + } + std::thread::sleep(Duration::from_millis(100)); + } + // Never came up (e.g. a toolchain/SQLite issue) or died. let _ = child.kill(); let _ = child.wait(); - return None; + let _ = std::fs::remove_file(&db); } - - Some(Self { - host: "127.0.0.1".to_string(), - port, - child: Some(child), - db: Some(db), - _gate: gate, - }) + None } fn settings(&self, username: &str, password: &str) -> ClientSettings { @@ -327,8 +340,15 @@ fn a_chat_room_message_is_delivered_between_users() { alice.join_room(room).expect("alice joins room"); bob.join_room(room).expect("bob joins room"); - // Give both joins time to register on the server before speaking. - std::thread::sleep(Duration::from_millis(500)); + // Each is in once the server has sent them the member list; a line said + // before bob is in never reaches him. + let deadline = Instant::now() + Duration::from_secs(5); + while alice.room_members(room).is_empty() + || bob.room_members(room).is_empty() + { + assert!(Instant::now() < deadline, "both should get into the room"); + std::thread::sleep(Duration::from_millis(50)); + } let _ = alice.take_room_events(); let _ = bob.take_room_events(); @@ -526,7 +546,14 @@ fn open_control_and_await_queue( let mut p = connect_retry(&cfg.listen_addr, Duration::from_secs(5))?; p.set_read_timeout(Some(Duration::from_secs(10)))?; p.write_all(&peer_init_bytes(&cfg.peer_username, "P", 0))?; + // The client answers a user-info request through the peer it registered + // for this connection, so the answer means a download queues on it + // rather than dialling a peer the server has never heard of. + p.write_all( + &soulseek_rs::message::peer::build_user_info_request().get_buffer(), + )?; p.flush()?; + expect_code(&mut p, 16, Duration::from_secs(10))?; let _ = cfg.ready.send(()); loop { @@ -860,10 +887,6 @@ fn file_downloads_over_p_and_f( .recv_timeout(Duration::from_secs(5)) .expect("mock uploader P connection"); - // The listener registers an incoming peer under its plain username; give - // that registration a moment to complete before queuing the download. - std::thread::sleep(Duration::from_millis(1500)); - let (_download, status_rx) = client .download( filename.to_string(), @@ -943,7 +966,6 @@ fn cancel_mid_transfer(peer_username: &str, stall_after: Option) { ready_rx .recv_timeout(Duration::from_secs(5)) .expect("mock uploader P connection"); - std::thread::sleep(Duration::from_millis(1500)); let (_download, status_rx) = client .download( @@ -1043,7 +1065,6 @@ fn a_cancelled_queued_download_declines_the_peers_offer() { ready_rx .recv_timeout(Duration::from_secs(5)) .expect("mock uploader P connection"); - std::thread::sleep(Duration::from_millis(1500)); let download_dir = unique_download_dir(); let (_download, status_rx) = client @@ -1117,7 +1138,6 @@ fn refusal_fails_the_download_quickly( ready_rx .recv_timeout(Duration::from_secs(5)) .expect("mock uploader P connection"); - std::thread::sleep(Duration::from_millis(1500)); let download_dir = unique_download_dir(); let (_download, status_rx) = client @@ -1217,7 +1237,6 @@ fn an_interrupted_download_resumes_from_its_partial_file() { ready_rx .recv_timeout(Duration::from_secs(5)) .expect("mock uploader P connection"); - std::thread::sleep(Duration::from_millis(1500)); let (_download, status_rx) = client .download( @@ -2936,9 +2955,9 @@ fn a_joined_room_reports_its_members_statistics() { assert!(sharer.login().expect("sharer login")); sharer.join_room(room).expect("sharer joins room"); - // Let the server register the sharer's membership and share counts, so - // the observer's own join reply describes a room that already has them. - std::thread::sleep(Duration::from_millis(750)); + // The sharer's membership and share counts have to land first, so the + // observer's own join reply describes a room that already has them. + fence(&sharer, "e2e_stats_observer"); let mut observer = Client::with_settings(server.settings("e2e_stats_observer", "pw")); @@ -3019,9 +3038,9 @@ fn watching_a_user_returns_their_status_and_share_counts() { watcher.connect().expect("watcher connect"); assert!(watcher.login().expect("watcher login")); - // Let the subject's share counts reach the server before watching, so - // the reply describes a user it already knows the statistics for. - std::thread::sleep(Duration::from_millis(750)); + // The subject's share counts have to reach the server before watching, + // so the reply describes a user it already knows the statistics for. + fence(&subject, "e2e_watch_watcher"); watcher .watch_user("e2e_watch_subject") @@ -3352,7 +3371,6 @@ fn download_progress_reports_the_rate_actually_received() { ready_rx .recv_timeout(Duration::from_secs(5)) .expect("mock uploader P connection"); - std::thread::sleep(Duration::from_millis(1500)); let (_download, status_rx) = client .download( @@ -3796,6 +3814,18 @@ fn await_room_event( None } +/// Wait until the server has handled everything `client` has sent, including +/// requests it never answers: soulfind takes a connection's messages in order, +/// so the answer to a question about `other` asked now comes after them. +fn fence(client: &Client, other: &str) { + client.request_user_info(other).expect("ask about a user"); + let deadline = Instant::now() + Duration::from_secs(5); + while client.user_info(other).is_none() { + assert!(Instant::now() < deadline, "the server should answer"); + std::thread::sleep(Duration::from_millis(50)); + } +} + #[test] fn a_ticker_set_in_a_room_reaches_the_other_members() { use soulseek_rs::types::RoomEvent; @@ -3812,7 +3842,12 @@ fn a_ticker_set_in_a_room_reaches_the_other_members() { alice.join_room(room).expect("alice joins"); bob.join_room(room).expect("bob joins"); - std::thread::sleep(Duration::from_millis(500)); + // A ticker reaches only who is in the room when it is set. + let deadline = Instant::now() + Duration::from_secs(5); + while bob.room_members(room).is_empty() { + assert!(Instant::now() < deadline, "bob should get into the room"); + std::thread::sleep(Duration::from_millis(50)); + } let _ = bob.take_room_events(); alice @@ -3856,11 +3891,15 @@ fn a_client_joining_later_receives_the_whole_ticker_board() { alice.connect().expect("alice connect"); assert!(alice.login().expect("alice login")); alice.join_room(room).expect("alice joins"); - std::thread::sleep(Duration::from_millis(300)); alice .set_room_ticker(room, "standing message") .expect("set ticker"); - std::thread::sleep(Duration::from_millis(500)); + // The room, alice included, is told once the board holds it. + await_room_event(&alice, Duration::from_secs(5), |event| { + matches!(event, RoomEvent::TickerAdded { room: r, .. } if r == room) + .then_some(()) + }) + .expect("alice should see her ticker go up"); // Carol joins afterwards and must be handed the board that already exists. let mut carol = @@ -3901,7 +3940,7 @@ fn the_global_room_feed_carries_a_room_we_never_joined() { watcher.join_global_room().expect("join global room"); talker.join_room(room).expect("talker joins a room"); - std::thread::sleep(Duration::from_millis(500)); + fence(&watcher, "e2e_global_talk"); let _ = watcher.take_room_events(); let body = "spoken where nobody is watching"; @@ -3925,7 +3964,7 @@ fn the_global_room_feed_carries_a_room_we_never_joined() { // And leaving the feed stops it: the next message must not arrive. watcher.leave_global_room().expect("leave global room"); - std::thread::sleep(Duration::from_millis(500)); + fence(&watcher, "e2e_global_talk"); let _ = watcher.take_room_events(); talker .say_in_room(room, "after leaving") @@ -3991,7 +4030,16 @@ fn a_room_search_is_answered_by_the_rooms_members() { ); sharer.join_room(room).expect("sharer joins"); searcher.join_room(room).expect("searcher joins"); - std::thread::sleep(Duration::from_millis(750)); + // A room search goes to whoever is in the room when it is sent. + let deadline = Instant::now() + Duration::from_secs(5); + while !searcher + .room_members(room) + .iter() + .any(|member| member == "e2e_rs_sharer") + { + assert!(Instant::now() < deadline, "both should get into the room"); + std::thread::sleep(Duration::from_millis(50)); + } let query = "relic"; searcher.search_room(room, query).expect("room search"); @@ -4028,7 +4076,7 @@ fn shared_interests_make_two_users_similar() { alice .add_dislike("e2e_interest_muzak") .expect("alice dislikes something"); - std::thread::sleep(Duration::from_millis(500)); + fence(&bob, "e2e_like_alice"); // Who else likes this item (code 112). let deadline = Instant::now() + Duration::from_secs(5); @@ -4122,7 +4170,7 @@ fn recommendations_are_returned_for_our_interests() { alice.add_interest(shared).expect("alice likes shared"); bob.add_interest(shared).expect("bob likes shared"); bob.add_interest(other).expect("bob likes other"); - std::thread::sleep(Duration::from_millis(500)); + fence(&bob, "e2e_rec_alice"); let deadline = Instant::now() + Duration::from_secs(8); let mut recommended = Vec::new(); @@ -4256,7 +4304,23 @@ fn a_changed_password_is_what_the_next_login_needs() { client .change_password("second-pw") .expect("change password"); - std::thread::sleep(Duration::from_millis(500)); + // soulfind stores the new password once it has hashed it, and forgets the + // change if our session closes first. The session stays until a login + // with the new password shows it landed; a refused one leaves it alone. + let deadline = Instant::now() + Duration::from_secs(10); + loop { + let mut fresh = + Client::with_settings(server.settings(user, "second-pw")); + fresh.connect().expect("connect with the new password"); + if matches!(fresh.login(), Ok(true)) { + break; + } + assert!( + Instant::now() < deadline, + "the new password must be accepted" + ); + std::thread::sleep(Duration::from_millis(100)); + } drop(client); let mut stale = Client::with_settings(server.settings(user, "first-pw")); @@ -4265,14 +4329,6 @@ fn a_changed_password_is_what_the_next_login_needs() { !matches!(stale.login(), Ok(true)), "the old password must stop working" ); - drop(stale); - - let mut fresh = Client::with_settings(server.settings(user, "second-pw")); - fresh.connect().expect("connect with the new password"); - assert!( - fresh.login().expect("login with the new password"), - "the new password must be accepted" - ); } #[test] @@ -4436,6 +4492,11 @@ fn phrases_the_server_excludes_are_kept_for_our_replies() { // network (code 160). Nicotine+ applies them to the files it offers in a // search reply, not to the searches it sends, and so do we — this pins // that the announced list arrives and is kept for that use. + // It binds listeners, so it waits for the server tests: a port handed + // to a soulfind that is still starting is otherwise up for grabs. + let _gate = SERVER_GATE + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); let mut phrases = Message::new(); phrases .write_int32(160) @@ -4604,7 +4665,8 @@ fn a_private_room_is_owned_granted_and_revoked() { owner .join_private_room(room) .expect("owner creates the room"); - std::thread::sleep(Duration::from_millis(750)); + // An invite reaches only a user already accepting invitations. + fence(&guest, "e2e_priv_owner"); // The guest is invited, and hears about it (code 139). owner @@ -4723,7 +4785,7 @@ fn a_private_room_someone_else_owns_cannot_be_taken() { owner .join_private_room(room) .expect("owner creates the room"); - std::thread::sleep(Duration::from_millis(750)); + fence(&owner, "e2e_taken_outsider"); // An outsider asking for the same name must be refused (code 1003), not // handed the room. @@ -4776,7 +4838,7 @@ fn an_acknowledged_offline_message_is_not_delivered_twice() { sender .send_private_message("e2e_ack_recipient", body) .expect("send to an offline user"); - std::thread::sleep(Duration::from_millis(500)); + fence(&sender, "e2e_ack_recipient"); let received_once = |label: &str| -> bool { let mut client = @@ -4794,7 +4856,9 @@ fn an_acknowledged_offline_message_is_not_delivered_twice() { std::thread::sleep(Duration::from_millis(100)); } } - std::thread::sleep(Duration::from_millis(500)); + // The acknowledgement goes out ahead of the message it answers, so + // once the server has handled what followed, it has the ack too. + fence(&client, "e2e_ack_sender"); got }; @@ -5348,7 +5412,7 @@ fn a_member_can_give_up_a_private_room_and_an_owner_can_disband_it() { owner .join_private_room(room) .expect("owner creates the room"); - std::thread::sleep(Duration::from_millis(750)); + fence(&guest, "e2e_leave_owner"); owner .add_room_member(room, "e2e_leave_guest") .expect("owner invites the guest"); @@ -5459,6 +5523,11 @@ fn a_disconnected_client_gives_its_listener_port_back() { // next one has to be able to advertise the port it was configured with, // not fall back to an ephemeral one nobody was told about. No server is // needed — binding the listener happens in connect(), before the dial. + // It binds listeners, so it waits for the server tests: a port handed + // to a soulfind that is still starting is otherwise up for grabs. + let _gate = SERVER_GATE + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); let port = free_port().expect("a port to hold"); let settings = |port: u16| ClientSettings { username: "e2e_port_release".to_string(), diff --git a/soulseek-rs-lib/tests/resume_e2e.rs b/soulseek-rs-lib/tests/resume_e2e.rs index 7271c25..89ef307 100644 --- a/soulseek-rs-lib/tests/resume_e2e.rs +++ b/soulseek-rs-lib/tests/resume_e2e.rs @@ -379,18 +379,21 @@ fn a_live_download_pauses_and_resumes_on_the_same_connection() { matches!(status, DownloadStatus::Paused { .. }) }); - // Give a read already in flight one timeout cycle to observe the pause. - std::thread::sleep(Duration::from_millis(1_200)); + // The pause is checked before every read, so once it lands only the read + // already in flight can finish: one 64 KiB buffer at most. Queue more + // than that, which a transfer ignoring the pause would take all of. let stable_len = fs::metadata(&part_path).expect("paused partial").len(); - let queued_while_paused = 8 * 1024; + let in_flight = 64 * 1024; + let queued_while_paused = in_flight + 8 * 1024; peer.write_all(&content[first_chunk..first_chunk + queued_while_paused]) .expect("queue bytes while paused"); peer.flush().expect("flush bytes while paused"); std::thread::sleep(Duration::from_millis(300)); - assert_eq!( - fs::metadata(&part_path).expect("paused partial").len(), - stable_len, - "a paused transfer must stop consuming its socket" + let paused_len = fs::metadata(&part_path).expect("paused partial").len(); + assert!( + paused_len <= stable_len + in_flight as u64, + "a paused transfer must stop consuming its socket: {paused_len} \ + bytes after pausing at {stable_len}" ); assert!(subject.client.resume_download(&control.username, filename)); diff --git a/soulseek-rs/src/port_mapping/mod.rs b/soulseek-rs/src/port_mapping/mod.rs index 4787084..f8e67b6 100644 --- a/soulseek-rs/src/port_mapping/mod.rs +++ b/soulseek-rs/src/port_mapping/mod.rs @@ -418,7 +418,18 @@ mod tests { thread::spawn(move || { let mut buf = [0u8; 64]; // Answer many requests (initial map + cleanup unmap + any renews). - while let Ok((n, from)) = socket.recv_from(&mut buf) { + loop { + let (n, from) = match socket.recv_from(&mut buf) { + Ok(received) => received, + // Windows reports a reply that met an already-closed + // client socket on the next receive; the gateway is fine. + Err(e) + if e.kind() == std::io::ErrorKind::ConnectionReset => + { + continue; + } + Err(_) => break, + }; let request = buf[..n].to_vec(); let internal = u16::from_be_bytes([request[4], request[5]]); let lifetime = u32::from_be_bytes([ @@ -465,20 +476,21 @@ mod tests { ); // Dropping the mapper must remove the mapping (a lifetime-0 request). + // A map the gateway was slow to answer is sent again, and that resend + // can arrive first. drop(mapper); - let unmap_req = requests - .recv_timeout(StdDuration::from_secs(3)) - .expect("dropping the mapper should send an unmap request"); - assert_eq!( + let unmapped = std::iter::from_fn(|| { + requests.recv_timeout(StdDuration::from_secs(3)).ok() + }) + .any(|request| { u32::from_be_bytes([ - unmap_req[8], - unmap_req[9], - unmap_req[10], - unmap_req[11] - ]), - 0, - "cleanup should request lifetime 0" - ); + request[8], + request[9], + request[10], + request[11], + ]) == 0 + }); + assert!(unmapped, "cleanup should request lifetime 0"); } /// A minimal in-process UPnP IGD: an HTTP server that answers every SOAP diff --git a/soulseek-rs/tests/cli_e2e.rs b/soulseek-rs/tests/cli_e2e.rs index a65e3e4..efea975 100644 --- a/soulseek-rs/tests/cli_e2e.rs +++ b/soulseek-rs/tests/cli_e2e.rs @@ -884,21 +884,28 @@ fn split_session_flags(args: &[&str]) -> (Vec, Vec) { /// /// `--no-config` is dropped for the obvious reason, and `--quiet` with it: /// these commands report the schedule the *server* chose on stderr, which is -/// the only place a test can see it. +/// the only place a test can see it. Session flags go to the daemon, as in +/// [`cli`]. fn cli_with_config( server: &TestServer, user: &str, config: &Path, args: &[&str], ) -> Output { + let (session_flags, command_args) = match mode() { + Mode::Local => { + (Vec::new(), args.iter().map(|a| (*a).to_string()).collect()) + } + Mode::Daemon => split_session_flags(args), + }; let mut all: Vec = server - .args(user) + .args_with(user, &session_flags) .into_iter() .filter(|arg| arg != "--no-config" && arg != "--quiet") .collect(); all.push("--config".to_string()); all.push(config.to_str().expect("utf-8 path").to_string()); - all.extend(args.iter().map(|a| (*a).to_string())); + all.extend(command_args); let refs: Vec<&str> = all.iter().map(String::as_str).collect(); run(&refs) } @@ -1888,8 +1895,19 @@ fn room_listen_streams_what_is_said_in_the_room() { .spawn() .expect("the binary should run"); - // Give the listener time to join before saying anything. - std::thread::sleep(Duration::from_secs(3)); + // The room announces the listener once the server has it in; a line + // said before that never reaches it. + let deadline = Instant::now() + Duration::from_secs(30); + while !speaker.take_room_events().iter().any(|event| { + matches!( + event, + soulseek_rs::types::RoomEvent::UserJoined { room, username } + if room == "cli_e2e_lobby_b" && username == "cli_e2e_room_reader" + ) + }) { + assert!(Instant::now() < deadline, "the listener never joined"); + std::thread::sleep(Duration::from_millis(50)); + } speaker .say_in_room("cli_e2e_lobby_b", "streamed line") .expect("say"); @@ -1958,6 +1976,11 @@ fn a_private_message_reaches_its_recipient() { fn message_read_streams_incoming_private_messages() { let server = server_or_skip!(); let sender = server.client("cli_e2e_pm_sender", Vec::new()); + // soulfind drops a message to a name it has never seen but keeps one for + // a known user until they log in, so however long the reader takes to + // get online, the line below reaches it. Registered before the reader's + // own session exists: a later login as them would take its place. + drop(server.client("cli_e2e_pm_reader", Vec::new())); settle(); let mut args = server.args("cli_e2e_pm_reader"); diff --git a/soulseek-rs/tests/common/mod.rs b/soulseek-rs/tests/common/mod.rs index d942a9e..82232af 100644 --- a/soulseek-rs/tests/common/mod.rs +++ b/soulseek-rs/tests/common/mod.rs @@ -60,12 +60,16 @@ pub struct Soulfind { } impl Soulfind { + /// soulfind exits if its port is taken between `free_port()` handing it + /// out and soulfind binding it, so a start that fails gets a fresh port. pub fn start() -> Option { - let port = free_port()?; - let db = std::env::temp_dir().join(format!("soulfind-{port}.db")); - let _ = std::fs::remove_file(&db); - let child = Self::spawn(port, &db)?; - Some(Self { child, db, port }) + (0..3).find_map(|_| { + let port = free_port()?; + let db = std::env::temp_dir().join(format!("soulfind-{port}.db")); + let _ = std::fs::remove_file(&db); + let child = Self::spawn(port, &db)?; + Some(Self { child, db, port }) + }) } fn spawn(port: u16, db: &Path) -> Option { @@ -78,8 +82,8 @@ impl Soulfind { let deadline = Instant::now() + Duration::from_secs(10); while Instant::now() < deadline { - // Whoever answers on the port has to be this child: one that - // lost the port and died would otherwise pass as up. + // A child that has exited is never taken as up: soulfind exits + // when its port is gone, and whoever took it may answer instead. if child.try_wait().ok().flatten().is_some() { return None; }