diff --git a/src/app/mod.rs b/src/app/mod.rs index bfdc657..d606d23 100644 --- a/src/app/mod.rs +++ b/src/app/mod.rs @@ -529,6 +529,27 @@ pub struct AppState { } impl AppState { + fn reset_room_state(&mut self) { + self.clip_player.stop(); + self.peers.clear(); + self.audio_levels.clear(); + self.locally_muted.clear(); + self.chat_messages.clear(); + self.chat_input.clear(); + self.attachment_data.clear(); + self.image_handle_cache.clear(); + self.pending_saves.clear(); + self.pending_plays.clear(); + self.invalid_audio.clear(); + self.connecting.clear(); + self.ever_connected.clear(); + self.recording = false; + self.recording_started = None; + self.call_started = None; + self.mic_level = 0.0; + self.self_sharing = false; + } + fn custom_sound_path(&self, sound: Sound) -> &str { let opt = match sound { Sound::SelfJoin => &self.config.custom_sound_self_join, @@ -1044,6 +1065,7 @@ fn update(state: &mut AppState, message: AppMessage) -> Task { AppMessage::UiEventReceived(event) => { match event { UiEvent::RoomJoined { ticket, self_id } => { + state.reset_room_state(); // Remember this gathering for one-click rejoin (W7 P5). The // emitted ticket is the canonical room door (topic + member // addr + label); push_recent de-dupes by topic and persists. @@ -1065,29 +1087,18 @@ fn update(state: &mut AppState, message: AppMessage) -> Task { notify::play(Sound::SelfJoin, state.config.custom_sound_self_join.as_deref()); } UiEvent::RoomLeft => { - state.clip_player.stop(); + state.reset_room_state(); state.ticket = "".to_string(); - state.peers.clear(); - state.audio_levels.clear(); - state.locally_muted.clear(); - state.call_started = None; - state.recording = false; - state.recording_started = None; - state.chat_messages.clear(); - state.chat_input.clear(); - state.attachment_data.clear(); - state.image_handle_cache.clear(); - state.pending_saves.clear(); - state.pending_plays.clear(); - state.invalid_audio.clear(); - state.connecting.clear(); - state.ever_connected.clear(); state.status_message = "Ready to connect".to_string(); state.current_screen = Screen::Home; - state.mic_level = 0.0; - state.self_sharing = false; notify::play(Sound::SelfLeave, state.config.custom_sound_self_leave.as_deref()); } + UiEvent::RoomReset => { + state.reset_room_state(); + state.ticket = "".to_string(); + state.status_message = "Ready to connect".to_string(); + state.current_screen = Screen::Home; + } UiEvent::PeerJoined { id, state: peer_state } => { state.peers.insert(id, peer_state); notify::play(Sound::PeerJoin, state.config.custom_sound_peer_join.as_deref()); @@ -5769,10 +5780,82 @@ impl Program for Icon { mod tests { use super::{ format_duration, initial_window_position, reconnect_attempt_chime, reconnected_chime, - set_peer_gate_config, set_peer_volume_config, AppConfig, GateMeter, METER_MAX, + set_peer_gate_config, set_peer_volume_config, AppConfig, AppState, AttachmentState, + ChatEntry, GateMeter, METER_MAX, }; use iroh::SecretKey; + #[test] + fn reset_room_state_clears_all_room_scoped_state() { + let mut state = AppState::default(); + let peer = SecretKey::generate().public(); + let attachment_id = [9u8; 32]; + let now = std::time::Instant::now(); + + state.peers.insert(peer, crate::network::PeerState { + name: "Peer".to_string(), + is_muted: false, + addr: iroh::EndpointAddr::from(peer), + sharing: None, + avatar: crate::avatar::Avatar::default(), + game: None, + }); + state.audio_levels.insert(peer, 0.5); + state.locally_muted.insert(peer); + state.chat_messages.push(ChatEntry { + name: "Peer".to_string(), + text: "old room".to_string(), + mine: false, + from: Some(peer.to_string()), + attachment: None, + }); + state.chat_input = "draft".to_string(); + state.attachment_data.insert(attachment_id, AttachmentState::Ready(vec![1])); + state.image_handle_cache.insert( + attachment_id, + iced::widget::image::Handle::from_bytes(vec![1]), + ); + state.pending_saves.insert(attachment_id); + state.pending_plays.insert(attachment_id); + state.invalid_audio.insert(attachment_id); + state.connecting.insert(peer); + state.ever_connected.insert(peer); + state.recording = true; + state.recording_started = Some(now); + state.call_started = Some(now); + state.mic_level = 0.75; + state.self_sharing = true; + state.clip_status.lock().unwrap().playing_id = Some(attachment_id); + + state.reset_room_state(); + + assert!(state.peers.is_empty()); + assert!(state.audio_levels.is_empty()); + assert!(state.locally_muted.is_empty()); + assert!(state.chat_messages.is_empty()); + assert!(state.chat_input.is_empty()); + assert!(state.attachment_data.is_empty()); + assert!(state.image_handle_cache.is_empty()); + assert!(state.pending_saves.is_empty()); + assert!(state.pending_plays.is_empty()); + assert!(state.invalid_audio.is_empty()); + assert!(state.connecting.is_empty()); + assert!(state.ever_connected.is_empty()); + assert!(!state.recording); + assert!(state.recording_started.is_none()); + assert!(state.call_started.is_none()); + assert_eq!(state.mic_level, 0.0); + assert!(!state.self_sharing); + + for _ in 0..50 { + if crate::audio::clip_player::status_snapshot(&state.clip_status).playing_id.is_none() { + return; + } + std::thread::sleep(std::time::Duration::from_millis(2)); + } + panic!("clip player did not stop during room reset"); + } + #[test] fn peer_gate_persists_when_on_and_clears_when_off() { let mut config = AppConfig::default(); diff --git a/src/audio/echo_cancel.rs b/src/audio/echo_cancel.rs index a278c06..5dc97b6 100644 --- a/src/audio/echo_cancel.rs +++ b/src/audio/echo_cancel.rs @@ -34,6 +34,18 @@ const NODE_READY_TIMEOUT: Duration = Duration::from_secs(3); /// nodes never leak past the call that created them. pub struct EchoCancelGuard { module_index: String, + source_name: String, + sink_name: String, +} + +impl EchoCancelGuard { + pub fn source_name(&self) -> &str { + &self.source_name + } + + pub fn sink_name(&self) -> &str { + &self.sink_name + } } impl Drop for EchoCancelGuard { @@ -58,12 +70,16 @@ pub fn enable(real_source: Option<&str>, real_sink: Option<&str>) -> Result, real_sink: Option<&str>) -> Result().is_err() { return Err(format!("unexpected pactl output: {module_index:?}")); } - let guard = EchoCancelGuard { module_index }; + let guard = EchoCancelGuard { module_index, source_name, sink_name }; // The virtual nodes appear shortly after the module loads; wait for both so // the subsequent capture/playback streams can actually target them. If they // never show, drop the guard (unloads) and report failure. - if !wait_for_nodes() { + if !wait_for_nodes(guard.source_name(), guard.sink_name()) { return Err("echo-cancel virtual nodes did not appear in time".to_string()); } @@ -102,10 +118,10 @@ pub fn enable(real_source: Option<&str>, real_sink: Option<&str>) -> Result bool { +fn wait_for_nodes(source_name: &str, sink_name: &str) -> bool { let deadline = Instant::now() + NODE_READY_TIMEOUT; loop { - if node_present("sources", EC_SOURCE) && node_present("sinks", EC_SINK) { + if node_present("sources", source_name) && node_present("sinks", sink_name) { return true; } if Instant::now() >= deadline { @@ -126,8 +142,30 @@ fn node_present(kind: &str, name: &str) -> bool { .any(|line| line.split('\t').nth(1) == Some(name)) } -/// Unloads any leftover `module-echo-cancel` instance we previously created -/// (identified by our virtual node names in its argument string). Best-effort. +fn pid_from_ec_args(args: &str) -> Option { + let source_prefix = format!("source_name={EC_SOURCE}."); + args.split_whitespace() + .find_map(|arg| arg.strip_prefix(&source_prefix))? + .parse() + .ok() +} + +fn ec_module_is_stale(args: &str, is_alive: impl Fn(u32) -> bool) -> bool { + pid_from_ec_args(args).is_some_and(|pid| !is_alive(pid)) +} + +#[cfg(target_os = "linux")] +fn process_is_alive(pid: u32) -> bool { + std::path::Path::new("/proc").join(pid.to_string()).exists() +} + +#[cfg(not(target_os = "linux"))] +fn process_is_alive(_pid: u32) -> bool { + true +} + +/// Unloads leftover PeerSpeak `module-echo-cancel` instances only when their +/// owning process is gone. Best-effort and conservative on non-Linux platforms. fn unload_stale() { let Ok(out) = Command::new("pactl").arg("list").arg("modules").arg("short").output() else { return; @@ -137,7 +175,10 @@ fn unload_stale() { let index = cols.next().unwrap_or(""); let name = cols.next().unwrap_or(""); let args = cols.next().unwrap_or(""); - if name == "module-echo-cancel" && args.contains(EC_SOURCE) && index.parse::().is_ok() { + if name == "module-echo-cancel" + && ec_module_is_stale(args, process_is_alive) + && index.parse::().is_ok() + { let _ = Command::new("pactl").arg("unload-module").arg(index).output(); crate::log_msg(&format!("Echo cancel: cleaned up stale module {index}")); } @@ -155,12 +196,42 @@ mod tests { #[ignore] fn enable_creates_and_unloads_nodes() { let guard = enable(None, None).expect("module-echo-cancel should load"); - assert!(node_present("sources", EC_SOURCE), "cleaned source must exist"); - assert!(node_present("sinks", EC_SINK), "reference sink must exist"); + let source_name = guard.source_name().to_string(); + let sink_name = guard.sink_name().to_string(); + assert!(node_present("sources", &source_name), "cleaned source must exist"); + assert!(node_present("sinks", &sink_name), "reference sink must exist"); drop(guard); // Give pactl a moment to tear the nodes down. std::thread::sleep(Duration::from_millis(300)); - assert!(!node_present("sources", EC_SOURCE), "source must be gone after unload"); - assert!(!node_present("sinks", EC_SINK), "sink must be gone after unload"); + assert!(!node_present("sources", &source_name), "source must be gone after unload"); + assert!(!node_present("sinks", &sink_name), "sink must be gone after unload"); + } + + #[test] + fn parses_owner_pid_only_from_our_source_name() { + assert_eq!( + pid_from_ec_args( + "aec_method=webrtc source_name=peerspeak_echocancel_source.4242 sink_name=peerspeak_echocancel_sink.4242" + ), + Some(4242) + ); + assert_eq!(pid_from_ec_args("aec_method=webrtc"), None); + assert_eq!( + pid_from_ec_args("source_name=peerspeak_echocancel_source.not-a-pid"), + None + ); + assert_eq!(pid_from_ec_args("source_name=someone_elses_source.4242"), None); + } + + #[test] + fn stale_decision_keeps_live_and_foreign_modules() { + let ours = "source_name=peerspeak_echocancel_source.4242"; + assert!(!ec_module_is_stale(ours, |pid| pid == 4242)); + assert!(ec_module_is_stale(ours, |_| false)); + assert!(!ec_module_is_stale("source_name=foreign.4242", |_| false)); + assert!(!ec_module_is_stale( + "source_name=peerspeak_echocancel_source.malformed", + |_| false + )); } } diff --git a/src/core/messages.rs b/src/core/messages.rs index 5840f46..17c90a0 100644 --- a/src/core/messages.rs +++ b/src/core/messages.rs @@ -106,6 +106,9 @@ pub enum CoreCommand { pub enum UiEvent { RoomJoined { ticket: String, self_id: String }, RoomLeft, + /// Clear room-scoped UI state after a failed in-call room switch, without a + /// leave chime. The persistent identity remains unchanged. + RoomReset, PeerJoined { id: EndpointId, state: PeerState }, PeerLeft { id: EndpointId }, /// The fixed reconnect grace expired and bounded background gossip recovery diff --git a/src/core/mod.rs b/src/core/mod.rs index 96787cd..a801078 100644 --- a/src/core/mod.rs +++ b/src/core/mod.rs @@ -132,14 +132,14 @@ type GraceTimers = Arc>>; type KnownPeers = - Arc>>>; + Arc>>>; #[derive(Clone)] struct RecoveryContext { coordinator: RecoveryCoordinator, room_state: Arc, known_peers: KnownPeers, - ticket: String, + topic_id: [u8; 32], } impl RecoveryContext { @@ -147,7 +147,7 @@ impl RecoveryContext { self.known_peers .lock() .unwrap() - .get(&self.ticket) + .get(&self.topic_id) .and_then(|peers| peers.get(peer_id)) .cloned() } @@ -157,7 +157,7 @@ impl RecoveryContext { } fn forget(&self, peer_id: EndpointId) { - if let Some(peers) = self.known_peers.lock().unwrap().get_mut(&self.ticket) { + if let Some(peers) = self.known_peers.lock().unwrap().get_mut(&self.topic_id) { peers.remove(&peer_id); } self.coordinator.cancel(peer_id); @@ -1207,6 +1207,7 @@ async fn run_core_loop( CoreCommand::Join { name, ticket, room_name, input_device, output_device, echo_cancellation, avatar } => { presence.name = name.clone(); presence.avatar = avatar; + let was_in_room = active_session.is_some(); // Finalize any recording before tearing down the old session — its // capture/mixer feeders are about to stop. @@ -1219,6 +1220,7 @@ async fn run_core_loop( session.shutdown(audio_backend.clone()).await; net.audio_router.clear(); net.file_router.clear(); + *current_room.lock().unwrap() = None; } // If a network-mode / identity change was deferred while a call was @@ -1268,6 +1270,17 @@ async fn run_core_loop( )); ticket_str }; + let topic_id = match PeerSpeakTicket::topic_of(&ticket_str) { + Some(topic_id) => topic_id, + None => { + crate::log_msg("Error invalid room ticket"); + if was_in_room { + let _ = ui_tx.send(UiEvent::RoomReset).await; + } + let _ = ui_tx.send(UiEvent::Error("invalid room ticket".to_string())).await; + continue; + } + }; // Per-session transport over the persistent endpoint, bound to the // persistent audio router so this call's inbound audio links route @@ -1292,14 +1305,14 @@ async fn run_core_loop( None, ); - // Snapshot THIS room's retained peers (by ticket) as extra bootstrap + // Snapshot THIS room's retained peers (by topic) as extra bootstrap // targets so a rejoin can dial them (A8) — including after a detour - // through another room, since the per-ticket archive isn't cleared. + // through another room, since the per-topic archive isn't cleared. // Resolution rides the persistent address book. let extra_bootstrap: Vec = known_peers .lock() .unwrap() - .get(&ticket_str) + .get(&topic_id) .map(|peers| peers.values().cloned().collect()) .unwrap_or_default(); @@ -1316,6 +1329,9 @@ async fn run_core_loop( )); if let Err(e) = room_state.join(&ticket_str, self_state.clone(), extra_bootstrap).await { crate::log_msg(&format!("Error room_state.join failed: {:?}", e)); + if was_in_room { + let _ = ui_tx.send(UiEvent::RoomReset).await; + } let _ = ui_tx.send(UiEvent::Error(format!("Failed to join room: {}", e))).await; net.audio_router.clear(); net.file_router.clear(); @@ -1342,12 +1358,11 @@ async fn run_core_loop( output_device.as_deref(), ) { Ok(guard) => { + let source_name = guard.source_name().to_string(); + let sink_name = guard.sink_name().to_string(); echo_cancel_guard = Some(guard); crate::log_msg("Echo cancellation enabled"); - ( - Some(crate::audio::echo_cancel::EC_SOURCE.to_string()), - Some(crate::audio::echo_cancel::EC_SINK.to_string()), - ) + (Some(source_name), Some(sink_name)) } Err(e) => { crate::log_msg(&format!( @@ -1368,6 +1383,9 @@ async fn run_core_loop( let (capture_target, playback_target) = (input_device.clone(), output_device.clone()); if let Err(e) = audio_backend.start_capture(capture_tx, capture_target) { + if was_in_room { + let _ = ui_tx.send(UiEvent::RoomReset).await; + } let _ = ui_tx.send(UiEvent::Error(format!("Failed to start capture: {}", e))).await; let _ = room_state.leave().await; net.audio_router.clear(); @@ -1380,6 +1398,9 @@ async fn run_core_loop( // production to the hardware clock instead of a fixed timer. let ring_fill = Arc::new(AtomicUsize::new(0)); if let Err(e) = audio_backend.start_playback(playback_rx, playback_target, ring_fill.clone()) { + if was_in_room { + let _ = ui_tx.send(UiEvent::RoomReset).await; + } let _ = ui_tx.send(UiEvent::Error(format!("Failed to start playback: {}", e))).await; let _ = audio_backend.stop(); let _ = room_state.leave().await; @@ -1681,23 +1702,57 @@ async fn run_core_loop( // track in Both mode) one aligned frame per cycle; Mixed mode // writes the single blended file as before. if mt_active { - if let Some(mt) = multitrack_mixer.lock().unwrap().as_mut() { - let res = (|| -> std::io::Result<()> { + let write_err = multitrack_mixer + .lock() + .unwrap() + .as_mut() + .map(|mt| -> std::io::Result<()> { for (id, f) in &stems { mt.write_peer(*id, f)?; } mt.write_mix(&record_mix)?; mt.end_cycle() - })(); - if let Err(e) = res { - crate::log_msg(&format!("Multitrack write failed: {e}")); - } + }) + .transpose() + .err(); + if let Some(e) = write_err { + crate::log_msg(&format!("Multitrack write failed: {e}")); + stop_recording( + &recorder_mixer, + &is_recording_mixer, + &multitrack_mixer, + &is_multitrack_mixer, + &ui_tx_mixer, + ).await; + let _ = ui_tx_mixer + .send(UiEvent::Error(format!( + "Recording stopped — write failed: {e}" + ))) + .await; + } + } else if is_recording_mixer.load(Ordering::Relaxed) { + let write_err = recorder_mixer + .lock() + .unwrap() + .as_mut() + .map(|rec| rec.write_frame(&record_mix)) + .transpose() + .err(); + if let Some(e) = write_err { + crate::log_msg(&format!("Recording write failed: {e}")); + stop_recording( + &recorder_mixer, + &is_recording_mixer, + &multitrack_mixer, + &is_multitrack_mixer, + &ui_tx_mixer, + ).await; + let _ = ui_tx_mixer + .send(UiEvent::Error(format!( + "Recording stopped — write failed: {e}" + ))) + .await; } - } else if is_recording_mixer.load(Ordering::Relaxed) - && let Some(rec) = recorder_mixer.lock().unwrap().as_mut() - && let Err(e) = rec.write_frame(&record_mix) - { - crate::log_msg(&format!("Recording write failed: {e}")); } let frame_to_send = if is_deafened_clone.load(Ordering::Relaxed) { @@ -1724,6 +1779,9 @@ async fn run_core_loop( let mut room_events = match room_state.subscribe_events().await { Ok(rx) => rx, Err(e) => { + if was_in_room { + let _ = ui_tx.send(UiEvent::RoomReset).await; + } let _ = ui_tx.send(UiEvent::Error(format!("Failed to subscribe events: {}", e))).await; continue; } @@ -1738,16 +1796,16 @@ async fn run_core_loop( let multitrack_events = multitrack.clone(); let is_multitrack_events = is_multitrack.clone(); let known_peers_events = known_peers.clone(); - // The ticket of the room this event loop serves, so peer add/remove - // updates the right per-ticket bucket in `known_peers` (A8 archive). - let ticket_events = ticket_str.clone(); + // The topic of the room this event loop serves, so peer add/remove + // updates the right per-topic bucket in `known_peers` (A8 archive). + let room_topic = topic_id; let (recovery_coordinator, recovery_task) = RecoveryCoordinator::spawn(room_state.clone()); let recovery_context = RecoveryContext { coordinator: recovery_coordinator, room_state: room_state.clone(), known_peers: known_peers.clone(), - ticket: ticket_str.clone(), + topic_id, }; let recovery_events = recovery_context.clone(); // Friends store + ui sender, so a connected peer who is a friend has @@ -1784,12 +1842,12 @@ async fn run_core_loop( ) .await; } - // Retain this peer under this room's ticket as a + // Retain this peer under this room's topic as a // future rejoin bootstrap target (A8). known_peers_events .lock() .unwrap() - .entry(ticket_events.clone()) + .entry(room_topic) .or_default() .insert(peer_id, state.addr.clone()); // If a multitrack recording is live, give this peer @@ -1845,7 +1903,7 @@ async fn run_core_loop( known_peers_events .lock() .unwrap() - .entry(ticket_events.clone()) + .entry(room_topic) .or_default() .insert(peer_id, state.addr.clone()); let _ = ui_tx_events.send(UiEvent::PeerUpdated { id: peer_id, state }).await; @@ -1905,6 +1963,9 @@ async fn run_core_loop( let mut conn_events = match transport.subscribe_conn_events().await { Ok(rx) => rx, Err(e) => { + if was_in_room { + let _ = ui_tx.send(UiEvent::RoomReset).await; + } let _ = ui_tx.send(UiEvent::Error(format!("Failed to subscribe conn events: {}", e))).await; continue; } @@ -2515,10 +2576,51 @@ async fn run_core_loop( mod tests { use super::{ apply_volume, audio_datagram_len_ok, frame_level, mix_frames, mix_stereo_frames, - next_game_change, stereo_to_mono, MicLevelMeter, MAX_OPUS_PAYLOAD, - MIC_LEVEL_REPORT_SAMPLES, + next_game_change, stereo_to_mono, KnownPeers, MicLevelMeter, PeerSpeakTicket, + MAX_OPUS_PAYLOAD, MIC_LEVEL_REPORT_SAMPLES, }; + #[test] + fn retained_peers_are_shared_by_topic_across_restamped_tickets() { + let topic_id = [23u8; 32]; + let original_host = iroh::SecretKey::generate().public(); + let member_host = iroh::SecretKey::generate().public(); + let retained_peer = iroh::SecretKey::generate().public(); + let original = PeerSpeakTicket { + host_addr: iroh::EndpointAddr::from(original_host), + topic_id, + name: "Room".to_string(), + }.to_string(); + let restamped = PeerSpeakTicket { + host_addr: iroh::EndpointAddr::from(member_host), + topic_id, + name: "Room".to_string(), + }.to_string(); + assert_ne!(original, restamped); + + let original_topic = PeerSpeakTicket::topic_of(&original).unwrap(); + let restamped_topic = PeerSpeakTicket::topic_of(&restamped).unwrap(); + assert_eq!(original_topic, restamped_topic); + + let retained_addr = iroh::EndpointAddr::from(retained_peer); + let known_peers: KnownPeers = + std::sync::Arc::new(std::sync::Mutex::new(std::collections::HashMap::new())); + known_peers + .lock() + .unwrap() + .entry(original_topic) + .or_default() + .insert(retained_peer, retained_addr.clone()); + + let found = known_peers + .lock() + .unwrap() + .get(&restamped_topic) + .and_then(|peers| peers.get(&retained_peer)) + .cloned(); + assert_eq!(found, Some(retained_addr)); + } + /// A frame of constant amplitude with the given sample count. fn frame(amp: i16, len: usize) -> Vec { vec![amp; len]