core: route critical commands through a reliable unbounded channel (A15)
cargo-deny / cargo-deny (push) Has been cancelled
windows-build / windows-build (push) Has been cancelled

CoreController::send put every app->core command on a single bounded
depth-100 channel via try_send and discarded the result. iced slider
drags emit ~60-120 commands/sec, so a drag burst could transiently
saturate the queue exactly when the user hit mute / released PTT /
left a room, silently dropping that critical command and leaving the
mic hot -- a privacy/state mismatch.

Split the queue by drop-tolerance:
- A pure delivery_class(&CoreCommand) classifier in messages.rs maps the
  7 continuous audio sliders to BestEffort and every other (discrete,
  human-paced) command to Reliable. The match has no wildcard arm, so a
  new CoreCommand variant fails to compile until it is classified.
- CoreController now holds two senders: an unbounded reliable channel
  and the existing bounded(100) best-effort channel. send() routes by
  class; Reliable uses unbounded send (fails only if the core loop is
  dead), BestEffort keeps today's bounded try_send.
- run_core_loop takes both receivers and drains them with a biased
  select: reliable first, best-effort second, game-change third.

Unbounded is safe because the only machine-rate producer (slider drags)
stays on the bounded channel; Reliable commands are all human-paced.
command_sender() and the awaiting Shutdown path are unchanged.

Implemented by Codex (gpt-5.5), senior-reviewed and verified here:
454 lib tests pass, clippy --all-targets clean, release builds.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
2026-06-27 04:19:40 -04:00
co-authored by Claude Opus 4.8
parent e3ff778d5b
commit ebfc39de46
2 changed files with 156 additions and 13 deletions
+131
View File
@@ -109,6 +109,80 @@ pub enum CoreCommand {
SetGameProcessMap(std::collections::BTreeMap<String, String>), SetGameProcessMap(std::collections::BTreeMap<String, String>),
} }
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum DeliveryClass {
Reliable,
BestEffort,
}
/// Route a command by how bad it is to drop it. Discrete, human-paced user
/// actions are Reliable (must land). The only high-frequency commands are the
/// continuous audio sliders, where dropping intermediate values is harmless;
/// those are BestEffort.
pub fn delivery_class(cmd: &CoreCommand) -> DeliveryClass {
match cmd {
CoreCommand::SetPeerVolume(_, _)
| CoreCommand::SetPeerPan(_, _)
| CoreCommand::SetPeerGate(_, _)
| CoreCommand::SetPeerEq(_, _)
| CoreCommand::SetInputVolume(_)
| CoreCommand::SetOutputVolume(_)
| CoreCommand::SetNoiseGateThreshold(_) => DeliveryClass::BestEffort,
CoreCommand::Join {
name: _,
ticket: _,
room_name: _,
input_device: _,
output_device: _,
echo_cancellation: _,
avatar: _,
}
| CoreCommand::Leave
| CoreCommand::Shutdown
| CoreCommand::ToggleMute
| CoreCommand::SetAvatar(_)
| CoreCommand::ToggleDeafen
| CoreCommand::SetPttMode(_)
| CoreCommand::SetPttActive(_)
| CoreCommand::SetPeerMuted(_, _)
| CoreCommand::SetMicMonitor {
enabled: _,
input_device: _,
}
| CoreCommand::SetNetworkMode(_)
| CoreCommand::SetRecording(_)
| CoreCommand::SetRecordingMode(_)
| CoreCommand::SendChat(_)
| CoreCommand::SendChatFile {
text: _,
attachment: _,
data: _,
}
| CoreCommand::FetchAttachment {
from: _,
attachment: _,
}
| CoreCommand::SetPixelpassPath(_)
| CoreCommand::ListAudioApps
| CoreCommand::StartScreenShare { audio_app: _ }
| CoreCommand::StopScreenShare
| CoreCommand::ViewShare(_)
| CoreCommand::RegenerateIdentity
| CoreCommand::AddFriend {
id: _,
name: _,
addr: _,
}
| CoreCommand::RemoveFriend(_)
| CoreCommand::RenameFriend(_, _)
| CoreCommand::SetPresenceMode(_)
| CoreCommand::SetGamePresenceEnabled(_)
| CoreCommand::SetGameOverride(_)
| CoreCommand::SetGameProcessMap(_) => DeliveryClass::Reliable,
}
}
#[derive(Debug, Clone)] #[derive(Debug, Clone)]
pub enum UiEvent { pub enum UiEvent {
RoomJoined { ticket: String, self_id: String }, RoomJoined { ticket: String, self_id: String },
@@ -199,3 +273,60 @@ pub enum UiEvent {
ShutdownComplete, ShutdownComplete,
Error(String), Error(String),
} }
#[cfg(test)]
mod tests {
use super::{delivery_class, CoreCommand, DeliveryClass};
use crate::audio::eq::EqSettings;
use crate::presence::PresenceMode;
use iroh::{EndpointId, SecretKey};
fn endpoint_id() -> EndpointId {
SecretKey::generate().public()
}
#[test]
fn continuous_audio_controls_are_best_effort() {
let peer = endpoint_id();
let commands = [
CoreCommand::SetPeerVolume(peer, 0.7),
CoreCommand::SetPeerPan(peer, -0.2),
CoreCommand::SetPeerGate(peer, 0.1),
CoreCommand::SetPeerEq(peer, EqSettings::default()),
CoreCommand::SetInputVolume(0.8),
CoreCommand::SetOutputVolume(0.9),
CoreCommand::SetNoiseGateThreshold(0.02),
];
for cmd in commands {
assert_eq!(delivery_class(&cmd), DeliveryClass::BestEffort);
}
}
#[test]
fn discrete_user_actions_are_reliable() {
let peer = endpoint_id();
let commands = [
CoreCommand::ToggleMute,
CoreCommand::SetPttActive(false),
CoreCommand::Leave,
CoreCommand::RegenerateIdentity,
CoreCommand::Join {
name: "Peer".to_string(),
ticket: "create".to_string(),
room_name: "Room".to_string(),
input_device: None,
output_device: None,
echo_cancellation: true,
avatar: crate::avatar::Avatar::default(),
},
CoreCommand::SetPeerMuted(peer, true),
CoreCommand::SetPresenceMode(PresenceMode::Normal),
CoreCommand::SendChat("hello".to_string()),
];
for cmd in commands {
assert_eq!(delivery_class(&cmd), DeliveryClass::Reliable);
}
}
}
+25 -13
View File
@@ -11,7 +11,7 @@ use crate::network::{
iroh_impl::{IrohTransport, AudioRouter, FileRouter}, iroh_impl::{IrohTransport, AudioRouter, FileRouter},
gossip::IrohGossipState, gossip::IrohGossipState,
}; };
use crate::core::messages::{CoreCommand, UiEvent}; use crate::core::messages::{CoreCommand, DeliveryClass, UiEvent, delivery_class};
use crate::core::recovery::RecoveryCoordinator; use crate::core::recovery::RecoveryCoordinator;
use crate::config::{NetworkMode, RecordingMode}; use crate::config::{NetworkMode, RecordingMode};
@@ -26,38 +26,44 @@ use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::time::Duration; use std::time::Duration;
pub struct CoreController { pub struct CoreController {
cmd_tx: mpsc::Sender<CoreCommand>, reliable_tx: mpsc::UnboundedSender<CoreCommand>,
besteffort_tx: mpsc::Sender<CoreCommand>,
} }
impl CoreController { impl CoreController {
pub fn new(ui_tx: mpsc::Sender<UiEvent>) -> Self { pub fn new(ui_tx: mpsc::Sender<UiEvent>) -> Self {
let (cmd_tx, cmd_rx) = mpsc::channel(100); let (reliable_tx, reliable_rx) = mpsc::unbounded_channel();
let (besteffort_tx, besteffort_rx) = mpsc::channel(100);
std::thread::spawn(move || { std::thread::spawn(move || {
let rt = tokio::runtime::Runtime::new().expect("Failed to create Tokio runtime"); let rt = tokio::runtime::Runtime::new().expect("Failed to create Tokio runtime");
rt.block_on(async move { rt.block_on(async move {
crate::log_msg("Starting core network loop in dedicated Tokio runtime"); crate::log_msg("Starting core network loop in dedicated Tokio runtime");
if let Err(e) = run_core_loop(cmd_rx, ui_tx).await { if let Err(e) = run_core_loop(reliable_rx, besteffort_rx, ui_tx).await {
crate::log_msg(&format!("App core loop failed: {:?}", e)); crate::log_msg(&format!("App core loop failed: {:?}", e));
} }
}); });
}); });
Self { cmd_tx } Self { reliable_tx, besteffort_tx }
} }
/// Queue a command for the core loop, best-effort. Returns `true` if it was /// Queue a command for the core loop. Reliable commands only fail when the
/// accepted, `false` if the channel is full or closed. (We return a plain /// core loop is dead; best-effort slider commands keep today's bounded
/// bool rather than the channel's `Result` so the bulky `CoreCommand` isn't /// try-send behavior. (We return a plain bool rather than the channel's
/// carried back by value in every caller's error type.) /// `Result` so the bulky `CoreCommand` isn't carried back by value in every
/// caller's error type.)
pub fn send(&self, cmd: CoreCommand) -> bool { pub fn send(&self, cmd: CoreCommand) -> bool {
self.cmd_tx.try_send(cmd).is_ok() match delivery_class(&cmd) {
DeliveryClass::Reliable => self.reliable_tx.send(cmd).is_ok(),
DeliveryClass::BestEffort => self.besteffort_tx.try_send(cmd).is_ok(),
}
} }
/// Clone the command sender for asynchronous one-shot sends that should wait /// Clone the command sender for asynchronous one-shot sends that should wait
/// for channel capacity instead of failing immediately on a full queue. /// for channel capacity instead of failing immediately on a full queue.
pub fn command_sender(&self) -> mpsc::Sender<CoreCommand> { pub fn command_sender(&self) -> mpsc::Sender<CoreCommand> {
self.cmd_tx.clone() self.besteffort_tx.clone()
} }
} }
@@ -952,7 +958,8 @@ async fn probe_friends_once(
} }
async fn run_core_loop( async fn run_core_loop(
mut cmd_rx: mpsc::Receiver<CoreCommand>, mut reliable_rx: mpsc::UnboundedReceiver<CoreCommand>,
mut besteffort_rx: mpsc::Receiver<CoreCommand>,
ui_tx: mpsc::Sender<UiEvent>, ui_tx: mpsc::Sender<UiEvent>,
) -> Result<(), anyhow::Error> { ) -> Result<(), anyhow::Error> {
let memory_lookup = iroh::address_lookup::memory::MemoryLookup::new(); let memory_lookup = iroh::address_lookup::memory::MemoryLookup::new();
@@ -1155,7 +1162,12 @@ async fn run_core_loop(
ping_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); ping_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
loop { loop {
let cmd = tokio::select! { let cmd = tokio::select! {
maybe_cmd = cmd_rx.recv() => match maybe_cmd { biased;
maybe_cmd = reliable_rx.recv() => match maybe_cmd {
Some(cmd) => cmd,
None => break,
},
maybe_cmd = besteffort_rx.recv() => match maybe_cmd {
Some(cmd) => cmd, Some(cmd) => cmd,
None => break, None => break,
}, },