fix(core,audio,app): Tier B bug-sweep fixes (F-05, F-06, F-10, F-11)
Four confirmed P2 findings from the 2026-06-22 adversarial bug sweep.
None change the wire format / PeerState / GOSSIP_PROTO — all local.
- F-05: re-key the A8 rejoin archive (known_peers) and RecoveryContext
by topic_id ([u8; 32]) instead of the raw ticket string. A W7-
restamped member ticket shares the room's topic but not its string,
so a rejoin-from-Recents previously missed the retained bootstrap
bucket and dropped to an empty bootstrap — the exact dead-end A8
fixed. Topic is derived once via PeerSpeakTicket::topic_of in Join;
a malformed ticket now fails early and clean.
- F-06: an in-call Join no longer leaks the old room's peers/chat into
the new room, nor strands stale presence on a failed switch. Core
captures was_in_room, clears current_room at teardown, and emits a
new local UiEvent::RoomReset on every post-teardown failure path so
a failed switch lands idle on Home. The UI's room-scoped clearing is
factored into AppState::reset_room_state(), called by RoomLeft,
RoomReset, and at the top of RoomJoined — so a successful switch
clears+repopulates seamlessly on the Room screen (no Home bounce, no
leave chime).
- F-10: echo-cancel virtual nodes now get per-PID-unique names
(peerspeak_echocancel_{source,sink}.<pid>); the guard carries them
and core targets them instead of the fixed constants. unload_stale
only unloads our modules whose owner PID is dead (/proc check, cfg-
gated; conservative elsewhere), so enabling AEC in one instance can
no longer tear down another live instance's call. Pure
pid_from_ec_args / ec_module_is_stale seams.
- F-11: a recording write failure now stops recording atomically
(best-effort finalize via stop_recording + one UI Error) instead of
looping the error at ~50 Hz with silent data loss. Both mixer
branches release the recorder mutex before calling stop_recording to
avoid a self-deadlock on the non-reentrant std::Mutex.
407 lib tests pass (+4), clippy --all-targets clean, release build
green. Tests-green only; the rejoin (F-05), in-call switch (F-06),
two-instance AEC (F-10), and disk-full (F-11) paths need a real run.
Implemented by Codex, reviewed + gates re-run by senior.
Co-Authored-By: Codex <codex@openai.com>
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Codex
Claude Opus 4.8
+102
-19
@@ -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> {
|
||||
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<AppMessage> {
|
||||
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<AppMessage> 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();
|
||||
|
||||
+84
-13
@@ -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<Echo
|
||||
// don't stack duplicate modules / fight over the virtual node names.
|
||||
unload_stale();
|
||||
|
||||
let owner_pid = std::process::id();
|
||||
let source_name = format!("{EC_SOURCE}.{owner_pid}");
|
||||
let sink_name = format!("{EC_SINK}.{owner_pid}");
|
||||
|
||||
let mut cmd = Command::new("pactl");
|
||||
cmd.arg("load-module")
|
||||
.arg("module-echo-cancel")
|
||||
.arg("aec_method=webrtc")
|
||||
.arg(format!("source_name={EC_SOURCE}"))
|
||||
.arg(format!("sink_name={EC_SINK}"));
|
||||
.arg(format!("source_name={source_name}"))
|
||||
.arg(format!("sink_name={sink_name}"));
|
||||
if let Some(src) = real_source.filter(|s| !s.is_empty()) {
|
||||
cmd.arg(format!("source_master={src}"));
|
||||
}
|
||||
@@ -85,12 +101,12 @@ pub fn enable(real_source: Option<&str>, real_sink: Option<&str>) -> Result<Echo
|
||||
if module_index.parse::<u64>().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<Echo
|
||||
}
|
||||
|
||||
/// Polls until both virtual nodes exist or the timeout elapses.
|
||||
fn wait_for_nodes() -> 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<u32> {
|
||||
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::<u64>().is_ok() {
|
||||
if name == "module-echo-cancel"
|
||||
&& ec_module_is_stale(args, process_is_alive)
|
||||
&& index.parse::<u64>().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
|
||||
));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
+131
-29
@@ -132,14 +132,14 @@ type GraceTimers = Arc<std::sync::Mutex<HashMap<EndpointId, tokio::task::JoinHan
|
||||
type SeenConnected = Arc<std::sync::Mutex<HashSet<EndpointId>>>;
|
||||
|
||||
type KnownPeers =
|
||||
Arc<std::sync::Mutex<HashMap<String, HashMap<EndpointId, EndpointAddr>>>>;
|
||||
Arc<std::sync::Mutex<HashMap<[u8; 32], HashMap<EndpointId, EndpointAddr>>>>;
|
||||
|
||||
#[derive(Clone)]
|
||||
struct RecoveryContext {
|
||||
coordinator: RecoveryCoordinator,
|
||||
room_state: Arc<IrohGossipState>,
|
||||
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<EndpointAddr> = 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 {
|
||||
})
|
||||
.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 Some(rec) = recorder_mixer.lock().unwrap().as_mut()
|
||||
&& let Err(e) = rec.write_frame(&record_mix)
|
||||
{
|
||||
} 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;
|
||||
}
|
||||
}
|
||||
|
||||
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<i16> {
|
||||
vec![amp; len]
|
||||
|
||||
Reference in New Issue
Block a user