Files
peerspeak/src/core/mod.rs
T
molluskandClaude Fable 5 5b80a1a010 core: pull the ticket off presence before the reap wait on host fault
Gemini's merge-round review (P2, CERTAIN): the fault handler ran
`stop_host().await` first, and a host that merely closed stdout but
lives on — trapped SIGINT, wedged — makes that call burn the full 2 s
stop grace before the SIGKILL fallback. For that whole window the dead
share stayed advertised: peers could still click Watch on it, and the
sharer's own UI kept saying "sharing".

The handler now retires the share where the fault is decided, not where
the corpse is confirmed: presence ticket removal and ScreenShareStopped
are emitted before the reap wait, and only the explanatory error (which
carries the unconfirmed-reap caveat) waits for `stop_host`. The
Stopped-before-Error contract is unchanged and still gated.

StopScreenShare's identical reap-then-presence ordering predates S2 and
is deliberately left alone (user-initiated stop, lower stakes); recorded
as a follow-up note instead of churning reviewed main-line code.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-31 14:51:59 -04:00

4307 lines
197 KiB
Rust

pub mod chatroster;
pub mod connstats;
pub mod fetchbudget;
pub mod jitter;
pub mod messages;
mod recovery;
mod teardown;
use crate::audio::eq::{Eq, EqSettings};
use crate::audio::{AudioBackend, PlatformAudioBackend};
use crate::codec::{AudioEncoder, opus_impl::OpusEncoder};
use crate::core::jitter::{FRAME_SAMPLES, JitterBuffer};
use crate::core::messages::{
CoalesceKey, CoreCommand, DeliveryClass, UiEvent, coalesce_key, delivery_class,
};
use crate::core::recovery::RecoveryCoordinator;
use crate::network::{
ConnEvent, NetworkTransport, PeerSpeakTicket, RoomEvent, RoomState, SelfPresence,
gossip::IrohGossipState,
iroh_impl::{AudioRouter, FileRouter, IrohTransport},
};
use crate::audio::multitrack::MultitrackRecorder;
use crate::config::{AudioProfile, NetworkMode, RecordingMode};
use crate::presence::PresenceMode;
use iroh::{
Endpoint, EndpointAddr, EndpointId, RelayMode, SecretKey, endpoint::presets, protocol::Router,
};
use iroh_gossip::net::Gossip;
use std::collections::{HashMap, HashSet};
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::{Arc, Mutex as StdMutex};
use std::time::Duration;
use tokio::sync::{Mutex, mpsc};
type CoalesceStore = Arc<StdMutex<HashMap<CoalesceKey, CoreCommand>>>;
// Mixer -> playback-worker handoff. The playback ring itself targets three
// 20ms frames; allow at most two more in flight so worker lag applies
// backpressure before the ring can overshoot to its 200ms cap (A6).
const PLAYBACK_HANDOFF_QUEUE_FRAMES: usize = 2;
const PLAYBACK_HANDOFF_RETRY: Duration = Duration::from_millis(1);
pub struct CoreController {
reliable_tx: mpsc::UnboundedSender<CoreCommand>,
coalesce: CoalesceStore,
besteffort_wake_tx: mpsc::Sender<()>,
}
#[derive(Clone)]
pub struct CoreCommandSender {
reliable_tx: mpsc::UnboundedSender<CoreCommand>,
coalesce: CoalesceStore,
besteffort_wake_tx: mpsc::Sender<()>,
}
fn coalesce_insert(map: &mut HashMap<CoalesceKey, CoreCommand>, cmd: CoreCommand) -> bool {
let Some(key) = coalesce_key(&cmd) else {
return false;
};
map.insert(key, cmd);
true
}
fn coalesce_pop(map: &mut HashMap<CoalesceKey, CoreCommand>) -> Option<CoreCommand> {
let key = map.keys().next().cloned()?;
map.remove(&key)
}
impl CoreCommandSender {
pub fn send(&self, cmd: CoreCommand) -> std::future::Ready<Result<(), CoreCommand>> {
std::future::ready(match delivery_class(&cmd) {
DeliveryClass::Reliable => self.reliable_tx.send(cmd).map_err(|e| e.0),
DeliveryClass::BestEffort => {
let inserted = {
let mut map = self.coalesce.lock().unwrap();
coalesce_insert(&mut map, cmd)
};
debug_assert!(inserted);
let _ = self.besteffort_wake_tx.try_send(());
Ok(())
}
})
}
}
impl CoreController {
pub fn new(ui_tx: mpsc::Sender<UiEvent>) -> Self {
let (reliable_tx, reliable_rx) = mpsc::unbounded_channel();
let coalesce = Arc::new(StdMutex::new(HashMap::new()));
let (besteffort_wake_tx, besteffort_wake_rx) = mpsc::channel(1);
let loop_coalesce = coalesce.clone();
let loop_besteffort_wake_tx = besteffort_wake_tx.clone();
std::thread::spawn(move || {
let rt = tokio::runtime::Runtime::new().expect("Failed to create Tokio runtime");
rt.block_on(async move {
crate::log_msg("Starting core network loop in dedicated Tokio runtime");
if let Err(e) = run_core_loop(
reliable_rx,
loop_coalesce,
besteffort_wake_rx,
loop_besteffort_wake_tx,
ui_tx,
)
.await
{
crate::log_msg(&format!("App core loop failed: {:?}", e));
}
});
});
Self {
reliable_tx,
coalesce,
besteffort_wake_tx,
}
}
/// Queue a command for the core loop. Reliable commands only fail when the
/// core loop is dead; best-effort slider commands are coalesced by control and
/// never fail to record their latest value. (We return a plain bool rather
/// than the channel's `Result` so the bulky `CoreCommand` isn't carried back by
/// value in every caller's error type.)
pub fn send(&self, cmd: CoreCommand) -> bool {
match delivery_class(&cmd) {
DeliveryClass::Reliable => self.reliable_tx.send(cmd).is_ok(),
DeliveryClass::BestEffort => {
let inserted = {
let mut map = self.coalesce.lock().unwrap();
coalesce_insert(&mut map, cmd)
};
debug_assert!(inserted);
let _ = self.besteffort_wake_tx.try_send(());
true
}
}
}
/// Clone a command sender for asynchronous one-shot sends.
pub fn command_sender(&self) -> CoreCommandSender {
CoreCommandSender {
reliable_tx: self.reliable_tx.clone(),
coalesce: self.coalesce.clone(),
besteffort_wake_tx: self.besteffort_wake_tx.clone(),
}
}
}
/// How long a peer may stay "reconnecting" after a transient drop before we give
/// up and evict it. Comfortably past the QUIC idle timeout (~30s) so a genuine
/// reconnect has time to complete, but short enough that a crashed/departed peer
/// clears from the room promptly.
const RECONNECT_GRACE: Duration = Duration::from_secs(45);
/// Opus frames sent by our encoder are one 20 ms mono frame, normally far below
/// this. 4000 bytes still leaves room for large valid Opus packets (well above a
/// 48 kHz / 60 ms frame) while bounding malicious datagram copy/decode churn.
const MAX_OPUS_PAYLOAD: usize = 4000;
/// If the Discoverable time-box tries to revert but discovery service reconfiguration
/// fails, retry soon while keeping the UI in the still-possible publishing state.
const DISCOVERY_REVERT_RETRY: Duration = Duration::from_secs(60);
fn audio_datagram_len_ok(len: usize) -> bool {
(4..=4 + MAX_OPUS_PAYLOAD).contains(&len)
}
async fn send_playback_frame(
tx: &std::sync::mpsc::SyncSender<Vec<i16>>,
mut frame: Vec<i16>,
) -> bool {
loop {
match tx.try_send(frame) {
Ok(()) => return true,
Err(std::sync::mpsc::TrySendError::Full(returned)) => {
frame = returned;
tokio::time::sleep(PLAYBACK_HANDOFF_RETRY).await;
}
Err(std::sync::mpsc::TrySendError::Disconnected(_)) => return false,
}
}
}
/// The presence label to broadcast for a detected game: its display name,
/// sanitized + length-capped, or `None` when there's no game or no broadcastable
/// name (a Steam appid without a manifest name, or a label that sanitizes empty).
/// Sanitizing here as well as at the gossip ingest boundary keeps the outgoing
/// value clean even though every peer re-sanitizes on receipt.
fn game_presence_label(game: Option<&crate::game::DetectedGame>) -> Option<String> {
game.and_then(|g| g.name.as_deref())
.map(crate::sanitize::sanitize_game_label)
.filter(|s| !s.is_empty())
}
/// Wait for the next game update. A closed sender permanently disables the
/// source by clearing the receiver; subsequent calls remain pending instead of
/// leaving an always-ready closed branch in the core `select!` loop.
async fn next_game_change(
game_rx: &mut Option<tokio::sync::watch::Receiver<Option<crate::game::DetectedGame>>>,
) -> Option<Option<crate::game::DetectedGame>> {
let Some(rx) = game_rx.as_mut() else {
return std::future::pending().await;
};
if rx.changed().await.is_err() {
*game_rx = None;
return None;
}
Some(rx.borrow_and_update().clone())
}
fn arm_discovery_retry(
discovery_deadline: &mut Option<tokio::time::Instant>,
now: tokio::time::Instant,
) {
let retry_deadline = now + DISCOVERY_REVERT_RETRY;
if discovery_deadline
.map(|current| current > retry_deadline)
.unwrap_or(true)
{
*discovery_deadline = Some(retry_deadline);
}
}
/// Per-peer reconnect grace timers (see [`RECONNECT_GRACE`]). Shared between the
/// room-event task (which arms one on a transient drop and cancels it on a
/// gossip rejoin) and the conn-event task (which cancels it when the audio link
/// actually comes back).
type GraceTimers = Arc<std::sync::Mutex<HashMap<EndpointId, tokio::task::JoinHandle<()>>>>;
/// Peers we've completed at least one audio link with. Lets the conn-event task
/// tell a genuine reconnect (arm an eviction timer) from a first-ever dial (don't).
/// Scrubbed whenever a peer is evicted or leaves so a later rejoin starts clean.
type SeenConnected = Arc<std::sync::Mutex<HashSet<EndpointId>>>;
type KnownPeers = Arc<std::sync::Mutex<HashMap<[u8; 32], HashMap<EndpointId, EndpointAddr>>>>;
/// Per-topic cap on the retained rejoin-bootstrap / recovery target table
/// (Tier C recovery-identity cap). Set comfortably above the live-roster cap
/// (`gossip::MAX_ACTIVE_PEERS`, 32) so a legitimate room — even one where every
/// member drops at once during a relay outage — never hits it, while an insider
/// who grace-cycles distinct identities (join, drop without a signed Leave,
/// repeat) cannot grow the table without bound. Combined with the recovery
/// terminal budget (which forgets a retained address when it gives up), abandoned
/// identities drain on their own, so this cap is a deterministic ceiling rather
/// than a pinnable slot pool.
const MAX_RETAINED_PEERS: usize = 64;
/// Whether a peer may be inserted into a retained-target table at `len` entries.
/// An update to an id already present is always allowed (it only refreshes an
/// address); a brand-new id is admitted only while below the cap. Mirrors the
/// gossip roster's `admit_into_roster` reject-when-full admission.
fn admit_retained(len: usize, is_new_id: bool, cap: usize) -> bool {
!is_new_id || len < cap
}
#[derive(Clone)]
struct RecoveryContext {
coordinator: RecoveryCoordinator,
room_state: Arc<IrohGossipState>,
known_peers: KnownPeers,
topic_id: [u8; 32],
}
impl RecoveryContext {
fn retained_addr(&self, peer_id: &EndpointId) -> Option<EndpointAddr> {
self.known_peers
.lock()
.unwrap()
.get(&self.topic_id)
.and_then(|peers| peers.get(peer_id))
.cloned()
}
fn cancel(&self, peer_id: EndpointId) {
self.coordinator.cancel(peer_id);
}
fn forget(&self, peer_id: EndpointId) {
if let Some(peers) = self.known_peers.lock().unwrap().get_mut(&self.topic_id) {
peers.remove(&peer_id);
}
self.coordinator.cancel(peer_id);
}
}
/// Cancel and forget a peer's pending grace timer, if any. No-op if none is armed.
fn cancel_grace_timer(timers: &GraceTimers, peer_id: &EndpointId) {
if let Some(handle) = timers.lock().unwrap().remove(peer_id) {
handle.abort();
}
}
/// Arm a per-peer reconnect grace timer that evicts the peer if its link hasn't
/// recovered within [`RECONNECT_GRACE`]. No-op if a timer is already pending for
/// the peer, so the earliest drop notice — whether the gossip `PeerConnectionLost`
/// or the transport `Connecting` — sets one hard deadline, rather than a flapping
/// link repeatedly resetting the clock and dodging eviction forever. On firing it
/// also scrubs the peer from `seen_connected` so a later rejoin isn't treated as a
/// reconnect on its initial dial.
struct GraceExpiry<'a> {
transport: &'a Arc<IrohTransport>,
jitter: &'a Arc<Mutex<HashMap<EndpointId, JitterBuffer>>>,
ui_tx: &'a mpsc::Sender<UiEvent>,
recovery: Option<&'a RecoveryContext>,
/// Terminal eviction also revokes the peer's chat authority (Phase 2):
/// the roster-bound name map entry goes with the peer.
chat_roster: &'a chatroster::ChatRoster,
}
fn arm_grace_timer(
timers: &GraceTimers,
seen_connected: &SeenConnected,
expiry: GraceExpiry<'_>,
grace: Duration,
peer_id: EndpointId,
) {
let mut timers_guard = timers.lock().unwrap();
if timers_guard.contains_key(&peer_id) {
return;
}
let transport_evict = expiry.transport.clone();
let jitter_evict = expiry.jitter.clone();
let ui_evict = expiry.ui_tx.clone();
let timers_evict = timers.clone();
let seen_evict = seen_connected.clone();
let recovery_evict = expiry.recovery.cloned();
let chat_roster_evict = expiry.chat_roster.clone();
let handle = tokio::spawn(async move {
tokio::time::sleep(grace).await;
crate::log_msg(&format!("Reconnect grace expired for peer {:?}", peer_id));
if let Some(recovery) = &recovery_evict
&& !recovery.coordinator.begin(peer_id)
{
return;
}
transport_evict.remove_audio_sender(peer_id);
// Terminal eviction revokes chat authority too (Phase 2): a readmission
// via fresh authenticated Announce re-registers the name on PeerJoined.
chat_roster_evict.remove(&peer_id);
if let Some(recovery) = &recovery_evict {
// Revoke roster authority before the first await in teardown. A
// verified Announce racing after this point is then a PeerJoined and
// cancels recovery instead of being erased after it was accepted.
recovery.room_state.mark_peer_disconnected(peer_id);
}
transport_evict.disconnect_peer(peer_id).await;
jitter_evict.lock().await.remove(&peer_id);
// Scrub our internal state *before* announcing the eviction, so anything
// that observes `PeerConnectionFailed` (or a rejoin racing it) sees a clean
// slate — a later dial for this identity is then a fresh first-dial, not a
// reconnect.
timers_evict.lock().unwrap().remove(&peer_id);
seen_evict.lock().unwrap().remove(&peer_id);
let Some(recovery) = recovery_evict else {
let _ = ui_evict
.send(UiEvent::PeerConnectionFailed { id: peer_id })
.await;
return;
};
if !recovery.coordinator.is_active(&peer_id) {
return;
}
let Some(addr) = recovery.retained_addr(&peer_id) else {
crate::log_msg(&format!(
"Cannot recover peer {:?}: no retained authenticated address",
peer_id
));
recovery.cancel(peer_id);
let _ = ui_evict
.send(UiEvent::PeerConnectionFailed { id: peer_id })
.await;
return;
};
match recovery.coordinator.activate(peer_id, addr) {
Ok(true) => {
let _ = ui_evict
.send(UiEvent::PeerRecoveryStarted { id: peer_id })
.await;
}
Ok(false) => {}
Err(()) => {
crate::log_msg(&format!(
"Cannot recover peer {:?}: recovery coordinator unavailable",
peer_id
));
let _ = ui_evict
.send(UiEvent::PeerConnectionFailed { id: peer_id })
.await;
}
}
});
timers_guard.insert(peer_id, handle);
}
/// Scale a frame in place by a per-peer volume factor, saturating to the i16
/// range. A volume within `f32::EPSILON` of 1.0 is treated as unity and skipped,
/// matching the mixer hot path that avoids touching unmodified frames.
fn apply_volume(frame: &mut [i16], vol: f32) {
if (vol - 1.0).abs() <= f32::EPSILON {
return;
}
for sample in frame.iter_mut() {
*sample = (*sample as f32 * vol).clamp(i16::MIN as f32, i16::MAX as f32) as i16;
}
}
/// Apply the listener's per-peer volume for the audio sender id currently being
/// mixed. The map key must be the same `EndpointId` used for the jitter buffer.
fn apply_peer_volume(frame: &mut [i16], peer_id: EndpointId, volumes: &HashMap<EndpointId, f32>) {
let vol = volumes.get(&peer_id).copied().unwrap_or(1.0);
apply_volume(frame, vol);
}
/// Normalized RMS level of a frame in `[0.0, 1.0]` (32768 = full scale), for the
/// UI level meter. An empty frame reads as 0.0.
fn frame_level(frame: &[i16]) -> f32 {
let sum_sq: f32 = frame.iter().map(|&x| (x as f32).powi(2)).sum();
let rms = (sum_sq / frame.len().max(1) as f32).sqrt();
(rms / 32768.0).clamp(0.0, 1.0)
}
/// Peak-hold every this many captured samples (~100ms @ 48kHz) before emitting a
/// [`UiEvent::MicLevel`], so the meter doesn't flood the UI runtime at frame rate.
const MIC_LEVEL_REPORT_SAMPLES: usize = 4800;
/// Peak-holds the raw mic level across captured frames and yields a value to
/// report roughly every [`MIC_LEVEL_REPORT_SAMPLES`] samples. Shared by the
/// in-call capture thread and the standalone monitor so both throttle and
/// peak-hold identically.
struct MicLevelMeter {
peak: f32,
acc: usize,
}
impl MicLevelMeter {
fn new() -> Self {
Self { peak: 0.0, acc: 0 }
}
/// Folds one frame into the running peak. Returns `Some(peak)` (and resets)
/// once enough samples have accumulated to emit a reading, else `None`.
fn push(&mut self, frame: &[i16]) -> Option<f32> {
self.peak = self.peak.max(frame_level(frame));
self.acc += frame.len();
if self.acc >= MIC_LEVEL_REPORT_SAMPLES {
let peak = self.peak;
self.peak = 0.0;
self.acc = 0;
Some(peak)
} else {
None
}
}
}
/// A standalone, capture-only mic monitor for gate calibration outside a call.
/// Owns the worker thread that reads raw PCM and reports its level; the PipeWire
/// capture stream itself lives in the shared backend. Tear down by stopping the
/// backend's capture (which closes the channel) and joining this thread.
struct MicMonitor {
thread: std::thread::JoinHandle<()>,
}
/// Drains a capture channel, reporting the raw (un-gated) mic level to the UI.
/// Returns when the channel closes (i.e. the backend's capture stream stopped).
fn run_mic_monitor(
rx: std::sync::mpsc::Receiver<Vec<i16>>,
ui_tx: mpsc::Sender<UiEvent>,
input_gain: Arc<std::sync::atomic::AtomicU32>,
) {
let mut meter = MicLevelMeter::new();
while let Ok(mut pcm) = rx.recv() {
// Mirror the in-call path: apply the input gain before metering so the
// test meter reflects the gained signal (and the input slider moves it).
apply_volume(&mut pcm, f32::from_bits(input_gain.load(Ordering::Relaxed)));
if let Some(peak) = meter.push(&pcm) {
// Drop on a full channel — a stale meter reading is harmless.
let _ = ui_tx.try_send(UiEvent::MicLevel(peak));
}
}
// Channel closed: the monitor was stopped. Snap the meter back to zero.
let _ = ui_tx.try_send(UiEvent::MicLevel(0.0));
}
/// Stops a standalone mic monitor if one is running. MUST NOT be called while a
/// room session is active — `backend.stop()` would also tear down the call's
/// capture/playback. Monitor and session are mutually exclusive by construction.
fn stop_mic_monitor(backend: &PlatformAudioBackend, monitor: Option<MicMonitor>) {
if let Some(m) = monitor {
let _ = backend.stop();
let _ = m.thread.join();
}
}
/// Sum per-peer frames sample-by-sample into one `frame_len`-sample bus, **without**
/// clamping — the lossless `i32` sum preserves the true peak so the mix-bus soft
/// limiter (see [`crate::audio::limiter`]) can ride it down to the ceiling instead
/// of the old hard clip shattering loud moments. Peers shorter than `frame_len`
/// contribute 0 past their end; an empty peer set yields a silent bus.
#[cfg(test)]
fn mix_frames(peer_frames: &[Vec<i16>], frame_len: usize) -> Vec<i32> {
let mut mixed = vec![0i32; frame_len];
for frame in peer_frames {
for (out, &sample) in mixed.iter_mut().zip(frame.iter()) {
*out += sample as i32;
}
}
mixed
}
/// Sum per-peer mono frames into one interleaved stereo `i32` bus. Center pan is
/// a special exact dual-mono path so the default listener mix is bit-for-bit the
/// old mono sum duplicated to both ears.
fn mix_stereo_frames(peer_frames: &[(Vec<i16>, f32)], frame_len: usize) -> Vec<i32> {
let mut mixed = vec![0i32; frame_len * crate::audio::PLAYBACK_CHANNELS];
for (frame, pan) in peer_frames {
if pan.abs() <= f32::EPSILON {
for (i, &sample) in frame.iter().take(frame_len).enumerate() {
let idx = i * crate::audio::PLAYBACK_CHANNELS;
let s = sample as i32;
mixed[idx] += s;
mixed[idx + 1] += s;
}
continue;
}
let (left_gain, right_gain) = crate::audio::pan::playback_pan_gains(*pan);
for (i, &sample) in frame.iter().take(frame_len).enumerate() {
let idx = i * crate::audio::PLAYBACK_CHANNELS;
let x = sample as f32;
mixed[idx] += (x * left_gain).round() as i32;
mixed[idx + 1] += (x * right_gain).round() as i32;
}
}
mixed
}
/// Fold an interleaved stereo frame to mono for the existing mixed WAV writers.
/// Center/default pan folds back to the exact old mono mix.
fn stereo_to_mono(stereo: &[i16]) -> Vec<i16> {
let mut mono = Vec::with_capacity(stereo.len() / crate::audio::PLAYBACK_CHANNELS);
for pair in stereo.chunks_exact(crate::audio::PLAYBACK_CHANNELS) {
let sum = pair[0] as i32 + pair[1] as i32;
mono.push((sum / 2).clamp(i16::MIN as i32, i16::MAX as i32) as i16);
}
mono
}
/// Handles the transport's per-peer link-state stream (`ConnEvent`): arms/cancels
/// reconnect grace timers, tracks which peers we've linked with, and forwards
/// link state to the UI. Pulled out of the conn-event task as a unit so the
/// reconnect-eviction behavior can be tested without standing up a full session.
/// Exposed (with a `grace` override) for that purpose; not part of the public API.
pub struct ConnEventHandler {
ui_tx: mpsc::Sender<UiEvent>,
grace_timers: GraceTimers,
seen_connected: SeenConnected,
transport: Arc<IrohTransport>,
jitter: Arc<Mutex<HashMap<EndpointId, JitterBuffer>>>,
recovery: Option<RecoveryContext>,
grace: Duration,
chat_roster: chatroster::ChatRoster,
}
impl ConnEventHandler {
pub fn new(
ui_tx: mpsc::Sender<UiEvent>,
grace_timers: GraceTimers,
seen_connected: SeenConnected,
transport: Arc<IrohTransport>,
jitter: Arc<Mutex<HashMap<EndpointId, JitterBuffer>>>,
) -> Self {
Self {
ui_tx,
grace_timers,
seen_connected,
transport,
jitter,
recovery: None,
grace: RECONNECT_GRACE,
chat_roster: chatroster::ChatRoster::default(),
}
}
/// Override the eviction grace window. For tests that can't wait 45s.
pub fn with_grace(mut self, grace: Duration) -> Self {
self.grace = grace;
self
}
/// Share the room's chat roster so a grace-expiry eviction fired from the
/// transport's link-state path also revokes chat authority (Phase 2).
pub fn with_chat_roster(mut self, chat_roster: chatroster::ChatRoster) -> Self {
self.chat_roster = chat_roster;
self
}
fn with_recovery(mut self, recovery: RecoveryContext) -> Self {
self.recovery = Some(recovery);
self
}
pub async fn handle(&self, event: ConnEvent) {
match event {
ConnEvent::Connecting(id) => {
// A reconnect (we've linked with this peer before): arm an eviction
// timer so a peer that never comes back is cleared even when gossip
// doesn't re-report the drop — the transport reliably re-emits this
// on every outage, gossip's NeighborDown does not. A first-ever dial
// (not yet in seen_connected) gets no timer; Connected cancels it on
// recovery.
if self.seen_connected.lock().unwrap().contains(&id) {
arm_grace_timer(
&self.grace_timers,
&self.seen_connected,
GraceExpiry {
transport: &self.transport,
jitter: &self.jitter,
ui_tx: &self.ui_tx,
recovery: self.recovery.as_ref(),
chat_roster: &self.chat_roster,
},
self.grace,
id,
);
}
let _ = self.ui_tx.send(UiEvent::PeerConnecting { id }).await;
}
ConnEvent::Connected(id) => {
// A transport event cannot readmit a grace-expired peer. Ignore a
// stale/racing link until authenticated gossip emits PeerJoined.
if self
.recovery
.as_ref()
.is_some_and(|recovery| recovery.coordinator.is_active(&id))
{
return;
}
// The audio link came back — the peer recovered within the grace
// window, so cancel its eviction.
cancel_grace_timer(&self.grace_timers, &id);
self.seen_connected.lock().unwrap().insert(id);
let _ = self.ui_tx.send(UiEvent::PeerConnected { id }).await;
}
ConnEvent::Left(id) => {
// The peer closed its link gracefully (intentional leave) — evict
// immediately, like a PeerLeft, instead of leaving it "reconnecting"
// until the grace timer or the slow gossip Leave.
cancel_grace_timer(&self.grace_timers, &id);
self.seen_connected.lock().unwrap().remove(&id);
if let Some(recovery) = &self.recovery {
recovery.forget(id);
}
self.transport.remove_audio_sender(id);
self.transport.disconnect_peer(id).await;
self.jitter.lock().await.remove(&id);
let _ = self.ui_tx.send(UiEvent::PeerLeft { id }).await;
}
}
}
}
struct ActiveSession {
room_state: Arc<IrohGossipState>,
capture_thread: std::thread::JoinHandle<()>,
datagram_task: tokio::task::JoinHandle<()>,
mixer_task: tokio::task::JoinHandle<()>,
event_task: tokio::task::JoinHandle<()>,
conn_event_task: tokio::task::JoinHandle<()>,
conn_stats_task: tokio::task::JoinHandle<()>,
recovery_task: tokio::task::JoinHandle<()>,
recovery_terminal_task: tokio::task::JoinHandle<()>,
grace_timers: GraceTimers,
transport: Arc<IrohTransport>,
/// The screen-share children and the echo-cancel module, held together
/// because their **destruction order** is load-bearing: the AEC module must
/// not unload while a pixelpass host is alive and fanning out (design v3.4
/// §7.1). `teardown` owns that ordering; see `core::teardown`.
teardown: SessionTeardown,
}
/// The session's teardown set, with the echo-cancel guard the platform actually
/// has. On non-Linux there is no AEC module, and `Infallible` makes that
/// structural — the `Option` cannot be `Some`.
#[cfg(target_os = "linux")]
type SessionTeardown = teardown::ScreenshareTeardown<
tokio::process::Child,
crate::audio::echo_cancel::EchoCancelGuard,
>;
#[cfg(not(target_os = "linux"))]
type SessionTeardown =
teardown::ScreenshareTeardown<tokio::process::Child, std::convert::Infallible>;
impl ActiveSession {
async fn shutdown(mut self, audio_backend: Arc<PlatformAudioBackend>) {
crate::log_msg("ActiveSession::shutdown started");
// Tear down any screen-share children first so the host stops streaming
// promptly, and so they are dead *and reaped* well before the AEC guard
// unloads at the end of this function (design v3.4 §7.1). Drop ordering
// is the backstop for the unwind path; this is the path we control.
self.teardown.shutdown_children().await;
self.datagram_task.abort();
self.mixer_task.abort();
self.event_task.abort();
self.conn_event_task.abort();
self.conn_stats_task.abort();
// Abort any pending reconnect grace timers so they can't fire a stray
// eviction (or touch a torn-down transport) after the session is gone.
for (_, handle) in self.grace_timers.lock().unwrap().drain() {
handle.abort();
}
self.recovery_task.abort();
self.recovery_terminal_task.abort();
crate::log_msg("Aborted tasks");
let audio_backend_clone = audio_backend.clone();
let _ = tokio::task::spawn_blocking(move || {
crate::log_msg("Stopping audio backend...");
let _ = audio_backend_clone.stop();
crate::log_msg("Audio backend stopped");
})
.await;
// Unload the echo-cancel module now that the audio streams releasing its
// virtual nodes have stopped. (Dropping the guard runs `pactl unload`.)
// The screen-share children were killed *and reaped* at the top of this
// function, so nothing pixelpass-side is alive to see the module vanish.
drop(self.teardown);
crate::log_msg("Leaving room...");
let _ = self.room_state.leave().await;
// Close peer links with the graceful goodbye code so remotes evict us
// promptly (not after the reconnect grace / slow gossip Leave), and stop
// our supervisors so none redial the about-to-close endpoint.
self.transport.leave().await;
crate::log_msg("Room left");
// NOTE: the router + endpoint are persistent (owned by the NetStack), so we
// deliberately do NOT shut them down here — only this session's peer links
// (closed by `transport.leave()` above) and tasks are torn down. The caller
// clears the persistent `audio_router` so stray inbound links are dropped.
crate::log_msg("Joining capture thread...");
let _ = self.capture_thread.join();
crate::log_msg("ActiveSession::shutdown complete");
}
}
/// The app's persistent network stack: ONE endpoint + gossip + router, built once
/// and kept alive for the whole `run_core_loop` lifetime (rebuilt only on a network
/// -mode or identity change). Room sessions come and go on top of this — `join`
/// subscribes a gossip topic + binds a per-session transport to `audio_router`,
/// `leave` drops them — but the endpoint stays reachable between calls. That's the
/// W7 prerequisite: the friends presence listener needs an endpoint up while idle,
/// and a node id can only have ONE live endpoint instance (proven by the dual-
/// endpoint spike — see `docs/contacts-plan.md` P4).
struct NetStack {
endpoint: Endpoint,
gossip: Gossip,
router: Router,
/// The persistent inbound-audio handler on `router`; per-join we bind the
/// active session's transport into it, and clear it on leave.
audio_router: AudioRouter,
/// The persistent chat-file-transfer handler on `router`; bound/cleared in
/// lock-step with `audio_router` (same session lifecycle).
file_router: FileRouter,
/// In-memory address book (ticket + gossip fed), shared with every session.
memory_lookup: iroh::address_lookup::memory::MemoryLookup,
}
impl NetStack {
async fn shutdown(self) {
crate::log_msg("Shutting down network stack (router + endpoint)...");
let _ = tokio::time::timeout(Duration::from_secs(1), self.router.shutdown()).await;
self.endpoint.close().await;
crate::log_msg("Network stack shut down");
}
}
/// Install the n0 DNS address-lookup services for a discovery `plan` (W7 P6), at
/// runtime, on an already-bound endpoint. The in-memory lookup (server-free, fed by
/// tickets + gossip) is always re-added; the n0 DNS *resolver* (`PkarrResolver` +
/// `DnsAddressLookup`, mirroring the `N0` preset) is added when `plan.resolver`; the
/// n0 DNS *publisher* (`PkarrPublisher`) when `plan.publisher`.
///
/// Idempotent and reversible: it builds the replacement services first, then clears
/// the service set and reinstalls exactly what the plan wants. Flipping `publisher`
/// off drops the publisher (its republish task ends when the last clone is dropped,
/// and the already-published record TTL-expires within ~30s) without an endpoint
/// rebuild and without disturbing resolution.
fn apply_discovery(
endpoint: &Endpoint,
memory_lookup: &iroh::address_lookup::memory::MemoryLookup,
plan: crate::discovery::LookupPlan,
) -> Result<(), anyhow::Error> {
use iroh::address_lookup::{
AddressLookupBuilder,
dns::DnsAddressLookup,
pkarr::{PkarrPublisher, PkarrResolver},
};
let services = endpoint.address_lookup()?;
let pkarr_resolver = if plan.resolver {
Some(PkarrResolver::n0_dns().into_address_lookup(endpoint)?)
} else {
None
};
let dns_resolver = if plan.resolver {
Some(DnsAddressLookup::n0_dns().into_address_lookup(endpoint)?)
} else {
None
};
let publisher = if plan.publisher {
Some(PkarrPublisher::n0_dns().into_address_lookup(endpoint)?)
} else {
None
};
services.clear();
// Always keep the local, server-free lookup (this is what ticket/gossip dialing
// depends on — it must survive every posture, including DirectOnly).
services.add(memory_lookup.clone());
if let Some(pkarr_resolver) = pkarr_resolver {
services.add(pkarr_resolver);
}
if let Some(dns_resolver) = dns_resolver {
services.add(dns_resolver);
}
if let Some(publisher) = publisher {
services.add(publisher);
}
Ok(())
}
/// Build the persistent network stack for the given identity + relay/discovery
/// posture. Binds the endpoint (the per-`NetworkMode` build hoisted out of the old
/// per-join path), spawns one gossip instance + one router accepting gossip and
/// audio, and kicks off `online()` in the background so app launch isn't blocked on
/// the relay handshake. `publish` is whether the presence posture is `Discoverable`
/// at build time (W7 P6) — it seeds the initial n0 DNS publish state.
async fn build_net_stack(
secret_key: SecretKey,
network_mode: NetworkMode,
memory_lookup: iroh::address_lookup::memory::MemoryLookup,
friends_handler: crate::presence_net::Handler,
publish: bool,
) -> Result<NetStack, anyhow::Error> {
// Bind with only the relay posture baked in (`Minimal` = crypto provider only,
// relay on/off per mode); n0 DNS discovery is installed uniformly below via
// `apply_discovery` so the Discoverable publish toggle is independent of the
// network mode and can be flipped later at runtime. The in-memory lookup is added
// here so dialing works even before the first `apply_discovery` (which re-adds it).
let relay_mode = match network_mode {
NetworkMode::DirectOnly => RelayMode::Disabled,
NetworkMode::RelayNoDiscovery | NetworkMode::N0Full => RelayMode::Default,
};
let endpoint = Endpoint::builder(presets::Minimal)
.secret_key(secret_key.clone())
.relay_mode(relay_mode)
.address_lookup(memory_lookup.clone())
.bind()
.await?;
// Install the n0 DNS services for the current (mode, publish) posture.
apply_discovery(
&endpoint,
&memory_lookup,
crate::discovery::lookup_plan(network_mode, publish),
)?;
// Bring the endpoint online in the background so launch isn't blocked on the
// relay handshake; joins/sends just work once it's up (online() is idempotent).
let ep = endpoint.clone();
tokio::spawn(async move { ep.online().await });
// One gossip instance for all rooms; sessions subscribe/unsubscribe topics on
// it. Frame budget raised above the 4 KB default so a hard-capped custom avatar
// (W4) can ride presence inline — all peers must use the same value.
let gossip = Gossip::builder()
.max_message_size(65536)
.spawn(endpoint.clone());
let audio_router = AudioRouter::new();
let file_router = FileRouter::new();
// The friends presence listener (W7 B2) rides this same persistent router as a
// third ALPN — it MUST be a handler here, not a standalone accept loop, since
// the router owns endpoint.accept(). Policy (who we answer / what room we
// report) is injected via `friends_handler`.
let router = Router::builder(endpoint.clone())
.accept(iroh_gossip::net::GOSSIP_ALPN, gossip.clone())
.accept(crate::protocol::AUDIO_ALPN, audio_router.clone())
.accept(crate::protocol::FILES_ALPN, file_router.clone())
.accept(
crate::presence_net::FRIENDS_ALPN,
crate::presence_net::FriendsProtocol::new(friends_handler),
)
.spawn();
Ok(NetStack {
endpoint,
gossip,
router,
audio_router,
file_router,
memory_lookup,
})
}
/// Retry policy for a live net-stack replacement, generic over the builder so
/// it is unit-testable without binding sockets: build for `requested`; if that
/// fails, build for `live` (the posture the old stack was actually running) so
/// a bad posture change degrades to the previous posture instead of leaving no
/// stack at all. When `requested == live` the second attempt is a plain retry.
///
/// `Ok((stack, mode, primary_err))` — a stack is up on `mode`; `primary_err`
/// is `Some` when the first attempt failed. `Err((primary, fallback))` — both
/// attempts failed and networking is gone.
async fn rebuild_with_fallback<T, E, F, Fut>(
mut build: F,
requested: NetworkMode,
live: NetworkMode,
) -> Result<(T, NetworkMode, Option<E>), (E, E)>
where
F: FnMut(NetworkMode) -> Fut,
Fut: std::future::Future<Output = Result<T, E>>,
{
match build(requested).await {
Ok(stack) => Ok((stack, requested, None)),
Err(primary) => match build(live).await {
Ok(stack) => Ok((stack, live, Some(primary))),
Err(fallback) => Err((primary, fallback)),
},
}
}
/// Tear down `old` and stand up a replacement stack for `requested_mode`.
///
/// A build failure here is rare (only the local socket bind can fail; the
/// relay handshake is backgrounded), but it used to propagate straight out of
/// `run_core_loop` with no `UiEvent`, silently killing every future command —
/// the app looked alive and did nothing. Instead, fall back to `live_mode`
/// via `rebuild_with_fallback`, tell the UI when the requested change did not
/// stick, and return the mode the new stack actually runs so the caller can
/// keep its state honest. `Err` only when both builds fail: networking is
/// gone (already reported to the UI as fatal) and the caller should exit.
#[allow(clippy::too_many_arguments)]
async fn replace_net_stack(
old: NetStack,
what: &str,
secret_key: &SecretKey,
requested_mode: NetworkMode,
live_mode: NetworkMode,
friends_handler: &crate::presence_net::Handler,
publish: bool,
ui_tx: &mpsc::Sender<UiEvent>,
) -> Result<(NetStack, NetworkMode), anyhow::Error> {
let lookup = old.memory_lookup.clone();
old.shutdown().await;
let outcome = rebuild_with_fallback(
|mode| {
build_net_stack(
secret_key.clone(),
mode,
lookup.clone(),
friends_handler.clone(),
publish,
)
},
requested_mode,
live_mode,
)
.await;
match outcome {
Ok((stack, mode, None)) => Ok((stack, mode)),
Ok((stack, mode, Some(primary))) => {
if mode == requested_mode {
// Same-posture retry succeeded — everything the user asked for
// is in effect, so log it rather than raising a UI error.
crate::log_msg(&format!(
"{what}: net stack build failed once ({primary:#}); retry succeeded"
));
} else {
let _ = ui_tx
.send(UiEvent::Error(format!(
"{what} failed ({primary:#}); staying on the previous \
network mode for this session"
)))
.await;
}
Ok((stack, mode))
}
Err((primary, fallback)) => {
let _ = ui_tx
.send(UiEvent::Error(format!(
"Networking lost: {primary:#} (recovery attempt also failed: \
{fallback:#}). Restart PeerSpeak to reconnect."
)))
.await;
Err(anyhow::anyhow!(
"net stack rebuild failed: {primary:#}; fallback: {fallback:#}"
))
}
}
}
/// Maximum number of *automatic* chat-attachment fetches in flight at once.
///
/// Auto-fetch (inline image preview) is triggered by an untrusted peer's chat
/// message, and each fetch is a detached task that can spend up to ~60s dialing
/// and reading. Without a bound, a room insider could spam attachment-carrying
/// chat to accumulate arbitrary pending tasks/dials (Tier C F-02). When the bound
/// is reached we simply skip the auto-fetch; the descriptor still renders and the
/// user can fetch it on demand (which is not rate-limited here).
const MAX_INFLIGHT_ATTACHMENT_FETCHES: usize = 4;
/// In-flight `(author, attachment_id)` markers for bounded, deduplicated auto-
/// fetches (Tier C F-02). Bounded by [`MAX_INFLIGHT_ATTACHMENT_FETCHES`].
type InflightAttachments = Arc<std::sync::Mutex<HashSet<(EndpointId, crate::files::AttachmentId)>>>;
/// Milliseconds since the Unix epoch — the time source handed to the
/// deterministic auto-fetch budget (mirrors the gossip plane's timestamps).
fn unix_now_ms() -> u64 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_millis() as u64)
.unwrap_or(0)
}
/// RAII bookkeeping for one bounded auto-fetch: holds the concurrency permit for
/// the task's lifetime and clears the in-flight `(author, id)` marker when the
/// fetch finishes (success OR failure), so the same image can be retried later.
struct AutoFetchGuard {
_permit: tokio::sync::OwnedSemaphorePermit,
inflight: InflightAttachments,
key: (EndpointId, crate::files::AttachmentId),
}
impl Drop for AutoFetchGuard {
fn drop(&mut self) {
self.inflight.lock().unwrap().remove(&self.key);
}
}
/// Whether to AUTO-fetch a chat image attachment. Only authenticated roster
/// authors qualify (closing the non-roster injection vector), only declared
/// sizes at or under [`crate::files::MAX_AUTO_IMAGE_BYTES`] (larger images get
/// a Load button instead, Phase 3B), and a `(author, id)` already being fetched
/// is skipped (dedup). The concurrency bound (permits) and the byte/request
/// budgets ([`fetchbudget::AutoFetchBudget`]) are enforced separately. Pure →
/// unit-testable (Tier C F-02).
fn should_auto_fetch(
is_image: bool,
author_in_roster: bool,
already_inflight: bool,
declared_size: u64,
) -> bool {
is_image
&& author_in_roster
&& !already_inflight
&& declared_size <= crate::files::MAX_AUTO_IMAGE_BYTES
}
/// Fetch a chat attachment's bytes from `from` over the file plane in a detached
/// task, then report the result to the UI via [`UiEvent::AttachmentReady`] /
/// [`UiEvent::AttachmentFailed`], tagged with `from` so the UI keys the bytes by
/// `(author, id)` and can't alias a same-id attachment from another sender. For images
/// (`is_image`) the bytes are defensively re-validated (decodable + within pixel
/// limits) before being handed to the renderer; an "Image" that doesn't decode is
/// reported as a failure rather than rendered. `guard` is `Some` for bounded
/// auto-fetches and `None` for user-initiated fetches; it is dropped when the
/// task ends, releasing the concurrency permit and the dedup marker.
fn spawn_attachment_fetch(
transport: Arc<IrohTransport>,
ui_tx: mpsc::Sender<UiEvent>,
from: EndpointId,
att: crate::files::ChatAttachment,
is_image: bool,
guard: Option<AutoFetchGuard>,
) {
tokio::spawn(async move {
// Held for the whole fetch; dropped here on completion (Tier C F-02).
let _guard = guard;
// Tell the UI a real fetch task exists for this key, so it can show a
// genuine loading state and dedup further clicks (Phase 3B). Sent from
// this task's channel handle, so it always precedes Ready/Failed.
let _ = ui_tx
.send(UiEvent::AttachmentFetchStarted { from, id: att.id })
.await;
match transport.fetch_attachment(from, &att).await {
Ok(data) => {
if is_image && crate::files::validate_image_bytes(&data).is_none() {
let _ = ui_tx
.send(UiEvent::AttachmentFailed {
from,
id: att.id,
error: "received image failed to decode".to_string(),
})
.await;
return;
}
let _ = ui_tx
.send(UiEvent::AttachmentReady {
from,
id: att.id,
data: Arc::new(data),
})
.await;
}
Err(e) => {
let _ = ui_tx
.send(UiEvent::AttachmentFailed {
from,
id: att.id,
error: e.to_string(),
})
.await;
}
}
});
}
fn spawn_music_fetch(
transport: Arc<IrohTransport>,
ui_tx: mpsc::Sender<UiEvent>,
from: EndpointId,
id: crate::files::AttachmentId,
size: u64,
) {
tokio::spawn(async move {
match transport.fetch_blob(from, id, size).await {
Ok(data) => {
let _ = ui_tx.send(UiEvent::MusicReady { from, id, data }).await;
}
Err(e) => {
let _ = ui_tx
.send(UiEvent::MusicFetchFailed {
from,
id,
error: e.to_string(),
})
.await;
}
}
});
}
fn spawn_music_prefetch(
transport: Arc<IrohTransport>,
ui_tx: mpsc::Sender<UiEvent>,
from: EndpointId,
id: crate::files::AttachmentId,
size: u64,
) {
tokio::spawn(async move {
match transport.fetch_blob(from, id, size).await {
Ok(data) => {
let _ = ui_tx
.send(UiEvent::MusicPrefetched { from, id, data })
.await;
}
Err(e) => {
let _ = ui_tx
.send(UiEvent::MusicFetchFailed {
from,
id,
error: e.to_string(),
})
.await;
}
}
});
}
/// Finalize and clear the active recording, if any, emitting `RecordingStopped`.
/// No-op when not recording. Called on stop, room leave, and room switch so a
/// recording is always closed cleanly (its WAV size fields patched).
async fn stop_recording(
recorder: &Arc<std::sync::Mutex<Option<crate::audio::recorder::Recorder>>>,
is_recording: &Arc<AtomicBool>,
multitrack: &Arc<std::sync::Mutex<Option<MultitrackRecorder>>>,
is_multitrack: &Arc<AtomicBool>,
ui_tx: &mpsc::Sender<UiEvent>,
) {
is_recording.store(false, Ordering::Relaxed);
is_multitrack.store(false, Ordering::Relaxed);
// Exactly one slot is ever active for a given recording, but finalize both
// defensively. The reported path is the file (mixed) or the session dir
// (multitrack).
let rec = recorder.lock().unwrap().take();
if let Some(rec) = rec {
let path = rec.path().to_string_lossy().to_string();
if let Err(e) = rec.finalize() {
crate::log_msg(&format!("Failed to finalize recording: {e}"));
}
crate::log_msg(&format!("Recording saved: {path}"));
let _ = ui_tx.send(UiEvent::RecordingStopped { path }).await;
}
let mt = multitrack.lock().unwrap().take();
if let Some(mt) = mt {
let path = mt.dir().to_string_lossy().to_string();
if let Err(e) = mt.finalize() {
crate::log_msg(&format!("Failed to finalize multitrack recording: {e}"));
}
crate::log_msg(&format!("Multitrack recording saved: {path}/"));
let _ = ui_tx.send(UiEvent::RecordingStopped { path }).await;
}
}
/// Persist the (core-owned) friends store and tell the GUI the new snapshot. Skips
/// the disk write when `read_only` (a malformed load — A16: don't clobber it) but
/// still emits so the UI reflects the in-memory change. Snapshots under the lock,
/// then releases it before the async send.
async fn persist_and_emit_friends(
friends: &Arc<std::sync::Mutex<crate::friends::FriendStore>>,
read_only: bool,
ui_tx: &mpsc::Sender<UiEvent>,
) {
let store = friends.lock().unwrap().clone();
if !read_only && let Err(e) = crate::friends::save(&store) {
crate::log_msg(&format!("friends: save failed: {e:#}"));
}
let _ = ui_tx
.send(UiEvent::FriendsUpdated {
friends: store.friends,
read_only,
})
.await;
}
/// How often the outbound presence scheduler refreshes friends' status. Each pass
/// opens one short connection per friend with a saved address, so the cost scales
/// with friend-count, not a fixed per-tick cost. 15s keeps the list feeling live
/// without aggressively probing peers for a best-effort signal; the manual Rescan
/// button covers the "update now" case below this interval.
const PING_INTERVAL: Duration = Duration::from_secs(15);
/// Delay before the FIRST presence pass, so the endpoint's background `online()`
/// has a moment to finish (otherwise the first probes fail and friends flash offline).
const PING_STARTUP_DELAY: Duration = Duration::from_secs(3);
/// One outbound presence-refresh pass (W7 B2): probe every friend and emit a
/// *definitive* status for each, so the UI self-heals every pass instead of only
/// ratcheting a friend upward. A friend with a saved address is probed and mapped
/// via [`crate::presence::presence_from_probe`] (a failed probe -> `Offline`); a
/// friend with no saved address (a bare add-by-id we've never met in a room) is
/// reported `Offline` directly, since a bare id can't resolve without discovery.
/// Probes run in parallel (friend counts are small). This is the fix for stale
/// "online"/"in a room" statuses lingering after a friend drops or leaves a room.
async fn probe_friends_once(
endpoint: Endpoint,
friends: crate::friends::FriendStore,
ui_tx: mpsc::Sender<UiEvent>,
) {
let mut set = tokio::task::JoinSet::new();
for f in friends.list() {
let id = f.id;
let Some(addr) = f.last_addr.clone() else {
// Nothing to dial yet — report Offline so a prior status can't stick.
let _ = ui_tx
.send(UiEvent::FriendPresence {
id,
presence: crate::presence::FriendPresence::Offline,
})
.await;
continue;
};
let ep = endpoint.clone();
set.spawn(async move {
let presence = match crate::presence_net::probe(&ep, addr).await {
Ok((from, reply)) => crate::presence::presence_from_probe(Some((&reply, from))),
Err(_) => crate::presence::presence_from_probe(None),
};
(id, presence)
});
}
while let Some(res) = set.join_next().await {
if let Ok((id, presence)) = res {
let _ = ui_tx.send(UiEvent::FriendPresence { id, presence }).await;
}
}
}
async fn run_core_loop(
mut reliable_rx: mpsc::UnboundedReceiver<CoreCommand>,
coalesce: CoalesceStore,
mut besteffort_wake_rx: mpsc::Receiver<()>,
besteffort_wake_tx: mpsc::Sender<()>,
ui_tx: mpsc::Sender<UiEvent>,
) -> Result<(), anyhow::Error> {
let memory_lookup = iroh::address_lookup::memory::MemoryLookup::new();
// Persistent identity (W7 P1): load a stable key so our node id survives
// launches — the foundation for the friends-first contacts model. Fall back
// to an ephemeral key only if the key file can't be read/created (e.g. no
// writable config dir), so a bad disk never blocks getting on a call. The
// `persisted` flag + error reason are surfaced to the UI (degraded state).
let (mut secret_key, mut identity_error) = match crate::identity::load_or_create() {
Ok(key) => (key, None),
Err(e) => {
let reason = format!("{e:#}");
crate::log_msg(&format!(
"identity: falling back to an ephemeral key (persistent load failed: {reason})"
));
(iroh::SecretKey::generate(), Some(reason))
}
};
// Tell the UI our node id + whether it's persisted. Re-sent after a regenerate.
let _ = ui_tx
.send(UiEvent::IdentityStatus {
node_id: secret_key.public().to_string(),
persisted: identity_error.is_none(),
error: identity_error.clone(),
})
.await;
// Peers seen per room, keyed by the room's join ticket, retained ACROSS leaves
// so a rejoin can bootstrap to them. This is the fix for A8: the room creator's
// own ticket lists only themselves as host, so on rejoin these retained peers
// are their only dial targets. Keyed by TICKET (not one flat set) so that
// revisiting a room after a detour through another one still remembers the
// first room's peers — the old single-set version cleared them on any ticket
// change, so an A→B→A bounce stranded the rejoiner with an empty bootstrap.
// Inner map keyed by peer id so updates refresh the address.
let known_peers: KnownPeers = Arc::new(std::sync::Mutex::new(HashMap::new()));
let audio_backend = Arc::new(PlatformAudioBackend::new());
let is_muted = Arc::new(AtomicBool::new(false));
let is_deafened = Arc::new(AtomicBool::new(false));
let ptt_mode = Arc::new(AtomicBool::new(false));
let ptt_active = Arc::new(AtomicBool::new(false));
let noise_gate_threshold = Arc::new(std::sync::atomic::AtomicU32::new(0.01f32.to_bits()));
// App-internal capture/playback gains (f32 bits), live-read by the audio loops.
let input_gain = Arc::new(std::sync::atomic::AtomicU32::new(1.0f32.to_bits()));
let output_gain = Arc::new(std::sync::atomic::AtomicU32::new(1.0f32.to_bits()));
// Opus encoder profile (W12) as a discriminant, live-read by the capture
// thread so a mid-call profile switch re-tunes the running encoder. Set from
// config via the GUI's startup `SetAudioProfile`; defaults to Balanced.
let audio_profile = Arc::new(std::sync::atomic::AtomicU8::new(
AudioProfile::default().as_u8(),
));
// Call recording: an optional live recorder (mic FIFO + WAV writer), shared
// by the capture thread (pushes mic) and the mixer task (writes mix frames).
// `is_recording` is a fast-path gate so the audio loops only take the lock
// while a recording is actually running.
let recorder: Arc<std::sync::Mutex<Option<crate::audio::recorder::Recorder>>> =
Arc::new(std::sync::Mutex::new(None));
let is_recording = Arc::new(AtomicBool::new(false));
// Multitrack (stem) recording: a parallel recorder used instead of the mixed
// `recorder` when `recording_mode` is Multitrack/Both. Exactly one of the two
// slots is ever active. `is_multitrack` is the fast-path gate the audio loops
// read (cheap) to decide whether to tap raw per-peer stems this cycle.
let multitrack: Arc<std::sync::Mutex<Option<MultitrackRecorder>>> =
Arc::new(std::sync::Mutex::new(None));
let is_multitrack = Arc::new(AtomicBool::new(false));
let mut recording_mode = RecordingMode::default();
let peer_volumes = Arc::new(Mutex::new(HashMap::<EndpointId, f32>::new()));
let peer_eq = Arc::new(Mutex::new(HashMap::<EndpointId, EqSettings>::new()));
let peer_pan = Arc::new(Mutex::new(HashMap::<EndpointId, f32>::new()));
// Listener-side per-peer noise-gate thresholds (normalized RMS, 0.0 = off).
let peer_gate = Arc::new(Mutex::new(HashMap::<EndpointId, f32>::new()));
// Peers locally muted by us: decoded for level metering but not mixed.
let locally_muted = Arc::new(Mutex::new(HashSet::<EndpointId>::new()));
// Sticky identity fields of our own presence (display name + W4 avatar), set on
// Join and changed via SetName/SetAvatar. Combined with the volatile per-announce
// fields (mute/addr/share ticket) by `SelfPresence::to_state` — the single place
// our `PeerState` is built. Defaults match the prior `current_name`/`current_avatar`.
let mut presence = SelfPresence {
name: "Anonymous".to_string(),
avatar: crate::avatar::Avatar::default(),
game: None,
music: None,
};
// Game detection (W17/W18): a background worker polls Steam state + the process
// list and publishes the debounced running game on a watch channel. Detection
// runs continuously (the GUI uses it for the local per-game background); whether
// the game is *broadcast* as presence is gated by `game_presence_enabled` (opt-in,
// seeded false; the GUI sends `SetGamePresenceEnabled` from config at startup).
// The override + process map start at their defaults and are set via commands.
let (game_detector, mut game_rx) = match crate::game::detector::GameDetector::spawn(
crate::game::ManualOverride::Auto,
std::collections::BTreeMap::new(),
) {
Ok(detector) => {
let rx = detector.subscribe();
(Some(detector), Some(rx))
}
Err(e) => {
crate::log_msg(&format!("game detector unavailable: {e}"));
(None, None)
}
};
let mut game_presence_enabled = false;
// The latest debounced detection, kept regardless of the broadcast toggle so a
// later opt-in can immediately publish whatever is currently running.
let mut current_game: Option<crate::game::DetectedGame> = None;
let mut network_mode = NetworkMode::default();
// Pixelpass binary override (config), and our own active screen share: the
// ticket rides our presence so the room — incl. late joiners — can watch,
// and the generation ties host-fault notices to this specific host spawn
// (see `ScreenShareHostFault`). One variable on purpose: the ticket and the
// generation must appear and vanish together, or a stale fault could tear
// down a share it doesn't belong to.
let mut pixelpass_override: Option<String> = None;
struct ActiveShare {
generation: u64,
ticket: String,
}
let mut current_sharing: Option<ActiveShare> = None;
// Monotonic per-spawn counter feeding `ActiveShare::generation`.
let mut share_generations: u64 = 0;
// Host faults re-enter the loop here (the notice-forwarder task can't touch
// loop state). The loop keeps `host_fault_tx` to clone into each share's
// forwarder, so this channel never closes — the select arm's `Some` pattern
// is total in practice and a closed-channel branch would be unreachable.
let (host_fault_tx, mut host_fault_rx) = mpsc::unbounded_channel::<u64>();
let mut active_session: Option<ActiveSession> = None;
// Standalone capture-only mic meter, live only when no session exists.
let mut mic_monitor: Option<MicMonitor> = None;
// Friends store (W7) — core now OWNS it (was the GUI's). Shared so the idle
// listener + ping scheduler read it. A malformed load yields an EMPTY store
// flagged read-only, so we never save over the damaged file (A16); the GUI
// surfaces the degraded state from `FriendsUpdated { read_only: true }`.
let (initial_friends, friends_read_only) = match crate::friends::load() {
Ok(store) => (store, false),
Err(e) => {
crate::log_msg(&format!(
"friends: load failed, starting read-only to avoid clobbering the file: {e:#}"
));
(crate::friends::FriendStore::default(), true)
}
};
let friends = Arc::new(std::sync::Mutex::new(initial_friends));
// Our presence posture, shared with the listener + scheduler. The GUI pushes it
// at startup + on change via SetPresenceMode; persistence stays in AppConfig.
let presence_mode = Arc::new(std::sync::Mutex::new(
crate::presence::PresenceMode::default(),
));
// The gathering we're currently in (its restamped member ticket + label), so the
// listener can offer friends a one-click Join. `None` when not in a call. Set on
// Join, cleared on Leave.
let current_room: Arc<std::sync::Mutex<Option<crate::presence::RoomPresence>>> =
Arc::new(std::sync::Mutex::new(None));
// Reply policy for the idle friends listener (B2): answer friends only, never
// while invisible (`should_answer`), and report our current gathering so a friend
// can one-click join. Reads the shared snapshots, so it stays correct as they
// change and survives a network-stack rebuild. Pure-sync (no awaits, no lock held
// across one). Built once and handed to every `build_net_stack`.
let friends_handler: crate::presence_net::Handler = {
let friends = friends.clone();
let presence_mode = presence_mode.clone();
let current_room = current_room.clone();
Arc::new(move |from| {
let mode = *presence_mode.lock().unwrap();
let allowed = crate::presence::should_answer(&from, &friends.lock().unwrap(), mode);
if !allowed {
return None;
}
let room = current_room.lock().unwrap().clone();
Some(crate::presence::ControlMsg::Pong { room })
})
};
// The persistent network stack (endpoint + gossip + router), built once at
// startup and kept alive for the app's lifetime. Room sessions ride on top of
// it (subscribe a topic + bind the audio router on join, clear on leave); it's
// rebuilt only when the network mode or identity changes. Moving `memory_lookup`
// in — all later access is via `net.memory_lookup`.
let initial_publish = presence_mode.lock().unwrap().publishes_to_discovery();
let mut net = match build_net_stack(
secret_key.clone(),
network_mode,
memory_lookup,
friends_handler.clone(),
initial_publish,
)
.await
{
Ok(stack) => stack,
Err(e) => {
// Only a local socket bind can fail here (the relay handshake is
// backgrounded), so this is fatal to networking — surface it rather
// than dying silently.
let _ = ui_tx
.send(UiEvent::Error(format!("Failed to start networking: {e}")))
.await;
return Err(e);
}
};
// Set when a network-mode / identity change arrives mid-call; the stack is
// rebuilt on the next Leave (or before the next Join), preserving the old
// "applies on next join" semantics while keeping the endpoint up while idle.
let mut net_rebuild_pending = false;
// The posture the live stack was actually built with. Trails `network_mode`
// while a rebuild is pending, and is the fallback posture when a rebuild
// fails (see `replace_net_stack`).
let mut net_mode = network_mode;
// When Discoverable is on, the instant it auto-reverts to Normal (W7 P6 time-box).
// `None` = not Discoverable, no pending revert. Set on SetPresenceMode(Discoverable),
// cleared on any other posture, and consumed by the select! branch below.
let mut discovery_deadline: Option<tokio::time::Instant> = None;
// Tell the GUI the loaded friends list (it renders from this, no longer owning
// it). Snapshot under the lock, then release it before the async send.
let initial_snapshot = friends.lock().unwrap().list().to_vec();
let _ = ui_tx
.send(UiEvent::FriendsUpdated {
friends: initial_snapshot,
read_only: friends_read_only,
})
.await;
// Drive the command loop AND the outbound presence ping scheduler together: the
// scheduler can't block command handling, so a probe pass runs in a detached task
// on each tick. `interval_at` delays the first pass so the endpoint can come
// online first.
let mut ping_interval = tokio::time::interval_at(
tokio::time::Instant::now() + PING_STARTUP_DELAY,
PING_INTERVAL,
);
ping_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
loop {
let cmd = tokio::select! {
biased;
maybe_cmd = reliable_rx.recv() => match maybe_cmd {
Some(cmd) => cmd,
// Every `CoreController`/`CoreCommandSender` is gone — the UI has
// dropped the core. Teardown happens once, after the loop.
None => break,
},
maybe_wake = besteffort_wake_rx.recv() => match maybe_wake {
Some(()) => {
let (cmd, has_more) = {
let mut map = coalesce.lock().unwrap();
let cmd = coalesce_pop(&mut map);
let has_more = !map.is_empty();
(cmd, has_more)
};
if has_more {
let _ = besteffort_wake_tx.try_send(());
}
match cmd {
Some(cmd) => cmd,
None => continue,
}
}
// ⚠️ UNREACHABLE BY CONSTRUCTION, twice over — do not mistake this
// for a live teardown path (phase 0b finding, 2026-07-26):
// 1. this function owns `besteffort_wake_tx` (cloned at the
// `CoreController::new` spawn site, used just above for the
// `has_more` re-arm), so the channel can never close while
// this loop is running;
// 2. even without that, every holder of a wake sender —
// `CoreController` and `CoreCommandSender` — holds
// `reliable_tx` too, and the `biased` select polls that one
// first, so the reliable arm always wins the race to exit.
// Teardown is hoisted after the loop, so if this arm is ever made
// reachable it is already covered — nothing to add here.
None => break,
},
// A share's notice-forwarder task reported the host's stdout ended.
// The `Some` pattern is total: this loop owns `host_fault_tx` (see
// its declaration), so the channel cannot close — no `None` arm is
// written because one would be unreachable by construction.
Some(generation) = host_fault_rx.recv() => {
CoreCommand::ScreenShareHostFault { generation }
}
game_change = next_game_change(&mut game_rx) => {
// The detector worker published a new debounced game (or `None`).
let Some(detected) = game_change else {
// Worker gone unexpectedly. The helper fused this source, so
// this logs once and the closed channel cannot spin select!.
crate::log_msg("game detector stopped; disabling game detection");
continue;
};
current_game = detected.clone();
// Always tell the GUI for the local per-game background + indicator.
let _ = ui_tx.send(UiEvent::GameChanged(detected.clone())).await;
// Broadcast as presence only when opted in; re-announce if in a room.
if game_presence_enabled {
presence.game = game_presence_label(detected.as_ref());
if let Some(session) = &active_session {
let self_state = presence.to_state(
is_muted.load(Ordering::Relaxed),
net.endpoint.addr(),
current_sharing.as_ref().map(|s| s.ticket.clone()),
);
let _ = session.room_state.update_self_state(self_state).await;
}
}
continue;
}
_ = ping_interval.tick() => {
// Fully dark while Invisible (the user's choice): don't even probe,
// so nothing we do touches a friend's machine. Otherwise refresh in a
// detached task capturing the CURRENT endpoint (a stack rebuild between
// ticks is naturally picked up next tick).
if *presence_mode.lock().unwrap() != crate::presence::PresenceMode::Invisible {
tokio::spawn(probe_friends_once(
net.endpoint.clone(),
friends.lock().unwrap().clone(),
ui_tx.clone(),
));
}
continue;
}
// W7 P6 time-box: Discoverable auto-reverts to Normal after DISCOVERY_TIMEBOX
// so a publish beacon never stands indefinitely. The branch is disabled
// (`if` guard) unless a deadline is armed; `unwrap_or_else` is unreachable
// belt-and-braces. On fire: stop publishing first, then commit Normal only
// if the endpoint's discovery services accepted the non-publishing plan.
_ = tokio::time::sleep_until(
discovery_deadline.unwrap_or_else(tokio::time::Instant::now),
), if discovery_deadline.is_some() => {
let previous_mode = *presence_mode.lock().unwrap();
if previous_mode != PresenceMode::Discoverable {
discovery_deadline = None;
continue;
}
let requested_mode = PresenceMode::Normal;
let now = tokio::time::Instant::now();
let plan = crate::discovery::lookup_plan(
network_mode,
requested_mode.publishes_to_discovery(),
);
let apply_result = apply_discovery(&net.endpoint, &net.memory_lookup, plan);
let (committed_mode, transition_error) =
crate::discovery::resolve_presence_transition(
previous_mode,
requested_mode,
apply_result.is_ok(),
);
*presence_mode.lock().unwrap() = committed_mode;
discovery_deadline = if committed_mode == PresenceMode::Discoverable {
Some(now + DISCOVERY_REVERT_RETRY)
} else {
None
};
match apply_result {
Ok(()) => {
crate::log_msg(
"discovery: Discoverable time-box elapsed → reverting to Normal",
);
let _ = ui_tx
.send(UiEvent::PresenceModeReverted {
mode: PresenceMode::Normal,
})
.await;
}
Err(e) => {
crate::log_msg(&format!("discovery: time-box revert failed: {e:#}"));
if committed_mode != requested_mode {
let _ = ui_tx
.send(UiEvent::PresenceModeReverted {
mode: committed_mode,
})
.await;
}
if let Some(message) = transition_error {
let _ = ui_tx
.send(UiEvent::Error(format!("{message} ({e:#})")))
.await;
}
}
}
continue;
}
};
match cmd {
CoreCommand::Shutdown => {
crate::log_msg("Core shutdown requested");
// Finalize recordings while capture/mixer feeders are still alive.
stop_recording(
&recorder,
&is_recording,
&multitrack,
&is_multitrack,
&ui_tx,
)
.await;
stop_mic_monitor(&audio_backend, mic_monitor.take());
if let Some(session) = active_session.take() {
session.shutdown(audio_backend.clone()).await;
net.audio_router.clear();
net.file_router.clear();
}
*current_room.lock().unwrap() = None;
net.shutdown().await;
let _ = ui_tx.send(UiEvent::ShutdownComplete).await;
break;
}
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.
stop_recording(
&recorder,
&is_recording,
&multitrack,
&is_multitrack,
&ui_tx,
)
.await;
// Clean up any existing session FIRST (this calls `transport.leave()`
// on the current endpoint), before any stack rebuild closes it.
if let Some(session) = active_session.take() {
crate::log_msg("Shutting down existing active session");
session.shutdown(audio_backend.clone()).await;
net.audio_router.clear();
net.file_router.clear();
*current_room.lock().unwrap() = None;
}
// Any advertised share died with that session — deliberately —
// so retire it HERE, before the invalid-ticket early exit below
// can skip it. Left populated, the killed host's stdout EOF
// would pass the ScreenShareHostFault staleness gate and
// surface as a spurious "ended unexpectedly" error on top of
// the ticket error (Gemini review of S2, P2-1).
current_sharing = None;
// If a network-mode / identity change was deferred while a call was
// active, rebuild the persistent stack now — after the old session is
// gone, before the new one binds — so this join uses the new posture.
if net_rebuild_pending {
let publish = presence_mode.lock().unwrap().publishes_to_discovery();
let (stack, live) = replace_net_stack(
net,
"Applying deferred network settings",
&secret_key,
network_mode,
net_mode,
&friends_handler,
publish,
&ui_tx,
)
.await?;
net = stack;
net_mode = live;
// If the new posture failed and we fell back, keep the mode
// state honest (and re-attemptable) rather than pretending
// the change applied. The join proceeds on the live stack.
network_mode = live;
net_rebuild_pending = false;
}
// Release a standalone mic monitor if running — it shares the
// backend's single capture stream, so it must stop before the
// call claims it. (Safe here: any session was just shut down.)
stop_mic_monitor(&audio_backend, mic_monitor.take());
// Use the app's persistent endpoint (bound once at startup); the
// per-join endpoint build is gone. A node id has exactly one live
// endpoint instance (W7 spike), so the friends listener and room
// audio necessarily share this one. online() is idempotent — it just
// ensures we're relay-reachable before joining the gossip swarm.
let endpoint = net.endpoint.clone();
endpoint.online().await;
// Determine target ticket
let ticket_str = if ticket.trim().is_empty() || ticket == "create" {
let topic_id: [u8; 32] = rand::random();
let host_addr = endpoint.addr();
crate::log_msg(&format!(
"Creating room. host_id={}, host_addrs={}, topic={}",
crate::short_id(&host_addr.id.to_string()),
host_addr.addrs.len(),
crate::short_bytes_hex(&topic_id)
));
// The creator's chosen cosmetic label rides in the ticket so
// every joiner inherits it; sanitize it before it leaves here.
let label = crate::sanitize::sanitize_name(&room_name);
let ticket = PeerSpeakTicket {
host_addr,
topic_id,
name: label,
};
ticket.to_string()
} else {
let ticket_str = ticket.trim().to_string();
crate::log_msg(&format!(
"Joining room with existing ticket={}",
crate::redact_for_log(&ticket_str)
));
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
// to it (cleared on leave). Gossip + router are persistent on the
// NetStack; the session just subscribes its topic below.
let transport = Arc::new(IrohTransport::new(endpoint.clone()));
net.audio_router.bind(&transport);
net.file_router.bind(&transport);
let room_state = Arc::new(IrohGossipState::new(
endpoint.clone(),
net.gossip.clone(),
net.memory_lookup.clone(),
secret_key.clone(),
));
// (The share was already retired beside the session teardown
// above; a fresh join starts not sharing.)
let self_state =
presence.to_state(is_muted.load(Ordering::Relaxed), endpoint.addr(), None);
// 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-topic archive isn't cleared.
// Resolution rides the persistent address book.
let extra_bootstrap: Vec<EndpointAddr> = known_peers
.lock()
.unwrap()
.get(&topic_id)
.map(|peers| peers.values().cloned().collect())
.unwrap_or_default();
let extra_bootstrap_ids = extra_bootstrap
.iter()
.map(|a| crate::short_id(&a.id.to_string()))
.collect::<Vec<_>>();
crate::log_msg(&format!(
"Attempting room_state.join self_id={}, self_name={:?}, sharing={}, extra_bootstrap={:?}",
crate::short_id(&self_state.addr.id.to_string()),
self_state.name,
self_state.sharing.is_some(),
extra_bootstrap_ids
));
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();
continue;
}
crate::log_msg("Joined room successfully via room_state");
// Setup raw audio channels
let (capture_tx, capture_rx) = std::sync::mpsc::channel();
let (playback_tx, playback_rx) =
std::sync::mpsc::sync_channel(PLAYBACK_HANDOFF_QUEUE_FRAMES);
// Echo cancellation: if enabled, load PipeWire's echo-cancel module
// bound to the chosen real devices and route capture/playback
// through its virtual nodes (the sink doubles as the AEC reference).
// The guard unloads the module on drop — including the early-return
// paths below, since it's a local until moved into the session. On
// any failure, warn and fall back to the direct devices.
#[cfg(target_os = "linux")]
let mut echo_cancel_guard = None;
#[cfg(target_os = "linux")]
let (capture_target, playback_target) = if echo_cancellation {
match crate::audio::echo_cancel::enable(
input_device.as_deref(),
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(source_name), Some(sink_name))
}
Err(e) => {
crate::log_msg(&format!(
"Echo cancellation unavailable, using direct devices: {e}"
));
let _ = ui_tx
.send(UiEvent::Error(format!(
"Echo cancellation unavailable: {e}"
)))
.await;
(input_device.clone(), output_device.clone())
}
}
} else {
(input_device.clone(), output_device.clone())
};
#[cfg(not(target_os = "linux"))]
let _ = echo_cancellation;
#[cfg(not(target_os = "linux"))]
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();
net.file_router.clear();
continue;
}
// Shared gauge: PipeWire publishes the playback ring's live depth
// here (drain side + fill side); the mixer reads it to pace
// 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;
net.audio_router.clear();
net.file_router.clear();
continue;
}
let jitter: Arc<Mutex<HashMap<EndpointId, JitterBuffer>>> =
Arc::new(Mutex::new(HashMap::new()));
// 1. Capture & encoding thread
let is_muted_clone = is_muted.clone();
let ptt_mode_clone = ptt_mode.clone();
let ptt_active_clone = ptt_active.clone();
let noise_gate_threshold_clone = noise_gate_threshold.clone();
let input_gain_clone = input_gain.clone();
let transport_clone = transport.clone();
let ui_tx_capture = ui_tx.clone();
let recorder_capture = recorder.clone();
let is_recording_capture = is_recording.clone();
let multitrack_capture = multitrack.clone();
let is_multitrack_capture = is_multitrack.clone();
let audio_profile_capture = audio_profile.clone();
let capture_thread = std::thread::spawn(move || {
use opus::{Application, Channels};
let mut encoder =
match OpusEncoder::new(48000, Channels::Mono, Application::Voip) {
Ok(enc) => enc,
Err(e) => {
crate::log_msg(&format!("Capture thread error: {:?}", e));
return;
}
};
// Tune the encoder to the configured profile (W12), then track
// the live discriminant so a mid-call switch re-applies it.
let mut current_profile =
AudioProfile::from_u8(audio_profile_capture.load(Ordering::Relaxed));
if let Err(e) = encoder.apply_profile(current_profile) {
crate::log_msg(&format!("Opus profile apply failed: {:?}", e));
}
// Per-sender packet sequence number, prepended to every frame so
// receivers can reorder and conceal loss. Wraps after ~years.
let mut seq: u32 = 0;
// Smooth noise gate (hysteresis + attack/release + hangover),
// carrying envelope state across frames. The live slider value
// is read per frame so changes apply immediately.
let mut gate = crate::audio::gate::NoiseGate::new(48000);
// Peak-held raw mic level for the settings meter, reported
// pre-gate/pre-mute so calibration reflects the true input.
let mut mic_meter = MicLevelMeter::new();
while let Ok(mut pcm) = capture_rx.recv() {
// Re-tune the encoder if the user switched profile mid-call.
// Cheap atomic load per frame; only reconfigures on change.
let want =
AudioProfile::from_u8(audio_profile_capture.load(Ordering::Relaxed));
if want != current_profile && encoder.apply_profile(want).is_ok() {
current_profile = want;
}
// Apply the input gain first so the meter, gate, and what we
// transmit all reflect the same (gained) signal.
apply_volume(
&mut pcm,
f32::from_bits(input_gain_clone.load(Ordering::Relaxed)),
);
if let Some(peak) = mic_meter.push(&pcm) {
let _ = ui_tx_capture.try_send(UiEvent::MicLevel(peak));
}
if is_muted_clone.load(Ordering::Relaxed) {
continue;
}
if ptt_mode_clone.load(Ordering::Relaxed)
&& !ptt_active_clone.load(Ordering::Relaxed)
{
continue;
}
let ng_bits = noise_gate_threshold_clone.load(Ordering::Relaxed);
let ng_thresh = f32::from_bits(ng_bits);
// Apply the gate in place; skip transmitting a fully-closed
// frame so we don't send pure silence (the receiver's jitter
// buffer conceals the gap).
if !gate.process(&mut pcm, ng_thresh) {
continue;
}
// Record what we transmit (post-gain, post-gate, post-mute):
// this is exactly the mic audio peers receive from us. The
// mixer task pairs it with the incoming mix.
if is_recording_capture.load(Ordering::Relaxed) {
if is_multitrack_capture.load(Ordering::Relaxed) {
if let Some(mt) = multitrack_capture.lock().unwrap().as_mut() {
mt.push_mic(&pcm);
}
} else if let Some(rec) = recorder_capture.lock().unwrap().as_mut() {
rec.push_mic(&pcm);
}
}
if let Ok(encoded) = encoder.encode(&pcm) {
// Frame on the wire: [seq: u32 LE][opus payload].
let mut packet = Vec::with_capacity(4 + encoded.len());
packet.extend_from_slice(&seq.to_le_bytes());
packet.extend_from_slice(&encoded);
seq = seq.wrapping_add(1);
transport_clone.broadcast(bytes::Bytes::from(packet));
}
}
});
// 2. Receiver task: parse the sequence header and hand each packet
// to that peer's jitter buffer. Decoding happens later, on the
// playout side, so loss can be concealed at the right moment.
let transport_recv = transport.clone();
let jitter_recv = jitter.clone();
let datagram_task = tokio::spawn(async move {
let mut datagram_rx = match transport_recv.receive_datagrams().await {
Ok(rx) => rx,
Err(e) => {
crate::log_msg(&format!("Receiver task error: {:?}", e));
return;
}
};
while let Some((from_peer, bytes)) = datagram_rx.recv().await {
if !transport_recv.audio_sender_admitted(from_peer) {
continue;
}
if !audio_datagram_len_ok(bytes.len()) {
// Malformed (< sequence header) or oversized Opus payload.
continue;
}
let seq = u32::from_le_bytes([bytes[0], bytes[1], bytes[2], bytes[3]]);
let payload = bytes[4..].to_vec();
let mut guard = jitter_recv.lock().await;
let buffer = match guard.entry(from_peer) {
std::collections::hash_map::Entry::Occupied(entry) => entry.into_mut(),
std::collections::hash_map::Entry::Vacant(entry) => {
match JitterBuffer::new() {
Ok(jb) => entry.insert(jb),
Err(e) => {
crate::log_msg(&format!(
"Failed to init jitter buffer for {:?}: {:?}",
from_peer, e
));
continue;
}
}
}
};
buffer.insert(seq, payload);
}
});
// 3. Mixing & level extraction loop task. Production is paced by
// the playback ring's fill level (the PipeWire hardware clock),
// NOT a fixed software timer: we produce a 20ms frame only when
// the ring is below its target depth, so the long-run mix rate
// auto-matches the device drain rate and the producer/consumer
// beat (which otherwise churns ~20% of audio) disappears. Each
// produced frame pulls one concealed frame per peer from its
// jitter buffer, applies per-peer volume, and sums.
let jitter_mixer = jitter.clone();
let is_deafened_clone = is_deafened.clone();
let peer_volumes_mixer = peer_volumes.clone();
let peer_eq_mixer = peer_eq.clone();
let peer_pan_mixer = peer_pan.clone();
let peer_gate_mixer = peer_gate.clone();
let locally_muted_mixer = locally_muted.clone();
let output_gain_mixer = output_gain.clone();
let ui_tx_mixer = ui_tx.clone();
let ring_fill_mixer = ring_fill.clone();
let recorder_mixer = recorder.clone();
let is_recording_mixer = is_recording.clone();
let multitrack_mixer = multitrack.clone();
let is_multitrack_mixer = is_multitrack.clone();
let mixer_task = tokio::spawn(async move {
// Mix-bus soft limiter: rides loud multi-peer moments down to
// the ceiling instead of hard-clipping. State carries across
// frames (see audio::limiter).
let mut limiter = crate::audio::limiter::SoftLimiter::new(48_000);
// Per-peer EQ filter state. Settings are live-cloned each
// cycle; state is rebuilt only when a peer's EQ changes.
let mut peer_eqs: HashMap<EndpointId, Eq> = HashMap::new();
// Per-peer noise-gate envelope state. The threshold is passed
// per frame (live slider), so the gate is never rebuilt — only
// created once per peer and dropped when the peer leaves.
let mut peer_noise_gates: HashMap<EndpointId, crate::audio::gate::NoiseGate> =
HashMap::new();
// When the ring is at/above target we have nothing to do; nap
// briefly and re-check. Short enough (relative to the ~60ms
// target and ~21ms device quantum) that we always refill well
// before the ring can run dry.
const IDLE_NAP: Duration = Duration::from_millis(2);
// Pushing a level event per frame floods the UI runtime at
// ~50/sec. We peak-hold per-peer levels across this many
// produced frames and emit once per window (~10/sec) —
// peak-hold so a brief transient still lights the indicator.
const LEVEL_EMIT_FRAMES: u32 = 5;
let mut level_peaks: HashMap<EndpointId, f32> = HashMap::new();
let mut frames_since_emit: u32 = 0;
loop {
// Pace to the hardware clock: only produce while the ring
// is draining below target. Otherwise yield and re-check.
if ring_fill_mixer.load(Ordering::Relaxed)
>= crate::audio::PLAYBACK_TARGET_SAMPLES
{
tokio::time::sleep(IDLE_NAP).await;
continue;
}
let current_volumes = peer_volumes_mixer.lock().await.clone();
let current_eq = peer_eq_mixer.lock().await.clone();
let current_pans = peer_pan_mixer.lock().await.clone();
let current_gates = peer_gate_mixer.lock().await.clone();
let muted_peers = locally_muted_mixer.lock().await.clone();
let mut peer_frames: Vec<(Vec<i16>, f32)> = Vec::new();
let mut peers_seen = HashSet::new();
// Multitrack stem capture: tap each peer's RAW decoded frame
// (pre-volume, pre-mute, pre-limiter) so the stems are clean
// source. Only collected while a multitrack recording is live.
let mt_active = is_recording_mixer.load(Ordering::Relaxed)
&& is_multitrack_mixer.load(Ordering::Relaxed);
let mut stems: Vec<(EndpointId, Vec<i16>)> = Vec::new();
{
let mut guard = jitter_mixer.lock().await;
for (&peer_id, buffer) in guard.iter_mut() {
// `None` means idle/buffering: contribute nothing,
// but keep a (zero) entry so the UI sees it idle.
let Some(mut frame) = buffer.pop_frame() else {
level_peaks.entry(peer_id).or_insert(0.0);
continue;
};
if mt_active {
stems.push((peer_id, frame.clone()));
}
// Listener-side per-peer noise gate, applied to the
// raw decoded frame (after the clean stem tap, before
// volume/EQ) so the threshold tracks the peer's true
// signal level regardless of our volume setting. The
// gate's "should transmit" return is irrelevant here —
// we only attenuate. Threshold 0 = off; the gate is
// created lazily and dropped when disabled.
let gate_threshold =
current_gates.get(&peer_id).copied().unwrap_or(0.0);
if gate_threshold > 0.0 {
peer_noise_gates
.entry(peer_id)
.or_insert_with(|| {
crate::audio::gate::NoiseGate::new(48_000)
})
.process(&mut frame, gate_threshold);
} else {
peer_noise_gates.remove(&peer_id);
}
apply_peer_volume(&mut frame, peer_id, &current_volumes);
let eq_settings = current_eq
.get(&peer_id)
.copied()
.unwrap_or_default()
.clamped();
if eq_settings.is_flat() {
peer_eqs.remove(&peer_id);
} else {
let needs_rebuild = peer_eqs
.get(&peer_id)
.map(|eq| eq.settings() != eq_settings)
.unwrap_or(true);
if needs_rebuild {
peer_eqs.insert(peer_id, Eq::new(eq_settings));
}
if let Some(eq) = peer_eqs.get_mut(&peer_id) {
eq.process_frame(&mut frame);
}
}
// Level is recorded even for locally-muted peers so
// the UI still shows that they're speaking.
let peak = level_peaks.entry(peer_id).or_insert(0.0);
*peak = peak.max(frame_level(&frame));
peers_seen.insert(peer_id);
// Locally muted: decoded above (jitter buffer advances,
// level shown) but not mixed into our output.
if muted_peers.contains(&peer_id) {
continue;
}
let pan = current_pans
.get(&peer_id)
.copied()
.unwrap_or(0.0)
.clamp(-1.0, 1.0);
peer_frames.push((frame, pan));
}
}
peer_eqs
.retain(|id, _| peers_seen.contains(id) || current_eq.contains_key(id));
peer_noise_gates.retain(|id, _| {
peers_seen.contains(id) || current_gates.contains_key(id)
});
// Lossless i32 sum, then the limiter applies the master
// output gain (in f32, so a boost past the ceiling is
// limited too) and rides peaks down to the ceiling.
let mixed_sum = mix_stereo_frames(&peer_frames, FRAME_SAMPLES);
let out_gain = f32::from_bits(output_gain_mixer.load(Ordering::Relaxed));
let mixed = limiter.process(&mixed_sum, out_gain);
let record_mix = stereo_to_mono(&mixed);
// Record the true call audio, independent of local deafen —
// deafen only silences our own monitor, not what the call
// carried. Multitrack writes raw per-peer stems (+ the mixed
// track in Both mode) one aligned frame per cycle; Mixed mode
// writes the single blended file as before.
if mt_active {
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()
})
.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;
}
}
let frame_to_send = if is_deafened_clone.load(Ordering::Relaxed) {
vec![0i16; mixed.len()]
} else {
mixed
};
if !send_playback_frame(&playback_tx, frame_to_send).await {
break;
}
// Emit coalesced peaks once per window, then reset.
frames_since_emit += 1;
if frames_since_emit >= LEVEL_EMIT_FRAMES {
let levels: Vec<(EndpointId, f32)> = level_peaks.drain().collect();
let _ = ui_tx_mixer.send(UiEvent::AudioLevels(levels)).await;
frames_since_emit = 0;
}
}
});
// 4. Room event subscriber task
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;
}
};
let ui_tx_events = ui_tx.clone();
let jitter_events = jitter.clone();
let transport_events = transport.clone();
let grace_timers: GraceTimers = Arc::new(std::sync::Mutex::new(HashMap::new()));
let grace_timers_events = grace_timers.clone();
let seen_connected: SeenConnected = Arc::new(std::sync::Mutex::new(HashSet::new()));
let seen_connected_events = seen_connected.clone();
let multitrack_events = multitrack.clone();
let is_multitrack_events = is_multitrack.clone();
let known_peers_events = known_peers.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, recovery_terminal_rx) =
RecoveryCoordinator::spawn(room_state.clone());
let recovery_context = RecoveryContext {
coordinator: recovery_coordinator,
room_state: room_state.clone(),
known_peers: known_peers.clone(),
topic_id,
};
let recovery_events = recovery_context.clone();
// Drain the recovery coordinator's terminal-eviction signals (Tier C
// recovery-identity cap). When background recovery exhausts its budget
// for a peer, forget its retained dial target so the per-topic retain
// table drains, scrub residual seen-connected state, and surface the
// failure. A peer that later returns can still rejoin via a gossip
// announce, so giving up never blocks a legitimate reconnect.
let recovery_terminal_ctx = recovery_context.clone();
let seen_connected_terminal = seen_connected.clone();
let ui_tx_terminal = ui_tx.clone();
let recovery_terminal_task = tokio::spawn(async move {
let mut terminal_rx = recovery_terminal_rx;
while let Some(peer_id) = terminal_rx.recv().await {
crate::log_msg(&format!(
"Background recovery gave up on peer {peer_id:?}; forgetting retained target"
));
recovery_terminal_ctx.forget(peer_id);
seen_connected_terminal.lock().unwrap().remove(&peer_id);
let _ = ui_tx_terminal
.send(UiEvent::PeerConnectionFailed { id: peer_id })
.await;
}
});
// Friends store + ui sender, so a connected peer who is a friend has
// their saved address auto-healed (W7) — populates `last_addr` so the
// presence scheduler can reach them later.
let friends_events = friends.clone();
let friends_read_only_events = friends_read_only;
// Bounded, deduplicated auto-fetch of chat image attachments (Tier C
// F-02): the permit pool caps concurrent fetch tasks; the in-flight
// set dedups identical (author, id) pairs.
let attachment_limiter =
Arc::new(tokio::sync::Semaphore::new(MAX_INFLIGHT_ATTACHMENT_FETCHES));
let inflight_attachments: InflightAttachments =
Arc::new(std::sync::Mutex::new(HashSet::new()));
// The authenticated chat roster for this room: id → roster-bound
// display name (chat-hardening Phase 2). Maintained from the same
// sequential event stream; shared because the detached grace-expiry
// timers (here and in the conn-event handler) must also revoke a
// terminally evicted peer's entry. Gates BOTH the chat text (only
// members render, under their roster name — never the wire name)
// and the automatic attachment fetch (Tier C F-02).
let chat_roster = chatroster::ChatRoster::default();
let chat_roster_events = chat_roster.clone();
let event_task = tokio::spawn(async move {
let roster = chat_roster_events;
// Byte/request budgets for automatic attachment fetches
// (Phase 3B). Only this sequential task consults it, so it
// needs no lock; time is passed in for testability.
let mut auto_fetch_budget = fetchbudget::AutoFetchBudget::new(unix_now_ms());
while let Some(event) = room_events.recv().await {
match event {
RoomEvent::PeerJoined(peer_id, state) => {
// A (re)join means the peer is back — cancel any
// pending reconnect grace timer before re-adding it.
roster.upsert(peer_id, &state.name);
cancel_grace_timer(&grace_timers_events, &peer_id);
recovery_events.cancel(peer_id);
transport_events.admit_audio_sender(peer_id);
// Establish the audio connection as soon as the peer
// is known (the transport dedupes the full-mesh race).
// Hand over the full address so reconnects can dial
// it directly rather than via the gossip lookup.
transport_events.connect_peer(state.addr.clone()).await;
// Auto-heal a friend's saved address (W7): if this
// peer is a friend, refresh their last_addr so the
// presence scheduler can reach them between rooms.
let healed = friends_events
.lock()
.unwrap()
.note_seen(&peer_id, state.addr.clone());
if healed {
persist_and_emit_friends(
&friends_events,
friends_read_only_events,
&ui_tx_events,
)
.await;
}
// Retain this peer under this room's topic as a
// future rejoin bootstrap target (A8), bounded by the
// per-topic retain cap (Tier C recovery-identity cap):
// refreshing a peer we already track is always allowed,
// a brand-new identity only while below the cap.
{
let mut kp = known_peers_events.lock().unwrap();
let bucket = kp.entry(room_topic).or_default();
let is_new_id = !bucket.contains_key(&peer_id);
if admit_retained(bucket.len(), is_new_id, MAX_RETAINED_PEERS) {
bucket.insert(peer_id, state.addr.clone());
} else {
crate::log_msg(&format!(
"Retain table full ({MAX_RETAINED_PEERS}); not retaining new peer {peer_id:?} for rejoin"
));
}
}
// If a multitrack recording is live, give this peer
// its own stem track (silence-padded back to t=0).
if is_multitrack_events.load(Ordering::Relaxed)
&& let Some(mt) = multitrack_events.lock().unwrap().as_mut()
&& let Err(e) = mt.add_peer(peer_id, &state.name)
{
crate::log_msg(&format!(
"multitrack add_peer (join) failed: {e}"
));
}
let _ = ui_tx_events
.send(UiEvent::PeerJoined { id: peer_id, state })
.await;
}
RoomEvent::PeerLeft(peer_id) => {
// Graceful leave — evict immediately (chat authority
// and roster-bound name included).
roster.remove(&peer_id);
cancel_grace_timer(&grace_timers_events, &peer_id);
seen_connected_events.lock().unwrap().remove(&peer_id);
// A signed Leave cancels background recovery and
// drops the retained target. Transient loss keeps it.
recovery_events.forget(peer_id);
transport_events.remove_audio_sender(peer_id);
transport_events.disconnect_peer(peer_id).await;
jitter_events.lock().await.remove(&peer_id);
let _ = ui_tx_events.send(UiEvent::PeerLeft { id: peer_id }).await;
}
RoomEvent::PeerUpdated(peer_id, state) => {
// Keep the roster-bound chat name current: a rename
// lands here as a state update (Phase 2). Future
// messages render under the new name; history keeps
// its stored snapshots.
roster.upsert(peer_id, &state.name);
// A re-announce means the peer is alive — cancel any
// pending grace timer. It may also carry a fresh
// address (peer back on a new network); refresh the
// retained dial target so a later reconnect re-reaches
// it. Idempotent: an ordinary mute/unmute update just
// re-records the same address.
cancel_grace_timer(&grace_timers_events, &peer_id);
recovery_events.cancel(peer_id);
transport_events.admit_audio_sender(peer_id);
transport_events.connect_peer(state.addr.clone()).await;
// Auto-heal a friend's saved address (W7) on the
// re-announce too — this is the path that catches a
// friend who moved networks mid-session.
let healed = friends_events
.lock()
.unwrap()
.note_seen(&peer_id, state.addr.clone());
if healed {
persist_and_emit_friends(
&friends_events,
friends_read_only_events,
&ui_tx_events,
)
.await;
}
// Refresh this room's retained rejoin target with the
// fresh addr (A8), under the per-topic retain cap. A
// re-announce from a peer we already track always
// refreshes; a new identity is bounded by the cap.
{
let mut kp = known_peers_events.lock().unwrap();
let bucket = kp.entry(room_topic).or_default();
let is_new_id = !bucket.contains_key(&peer_id);
if admit_retained(bucket.len(), is_new_id, MAX_RETAINED_PEERS) {
bucket.insert(peer_id, state.addr.clone());
} else {
crate::log_msg(&format!(
"Retain table full ({MAX_RETAINED_PEERS}); not retaining new peer {peer_id:?} for rejoin"
));
}
}
let _ = ui_tx_events
.send(UiEvent::PeerUpdated { id: peer_id, state })
.await;
}
RoomEvent::ChatMessage {
from,
// The wire name is sender-claimed and NEVER rendered:
// the roster-bound name below is the author label
// (chat-hardening Phase 2, the impersonation fix).
name: _,
text,
ts: _,
attachment,
} => {
// Final-authority roster gate: only a current
// authenticated member (including one inside its
// reconnect grace) may create chat UI work. The
// gossip loop's early known-author gate is defense
// in depth; this map is what actually decides.
let Some(name) = roster.name_of(&from) else {
crate::log_msg(&format!(
"Dropped chat from non-roster author {}",
crate::short_id(&from.to_string())
));
continue;
};
// Auto-fetch image attachments so they render inline
// without a click; non-image files wait for an explicit
// FetchAttachment (the "Save" chip). The descriptor was
// already filename-sanitized + size-capped on ingest.
//
// The auto path is an untrusted-peer-triggered detached
// task, so it is gated (Tier C F-02): only roster authors
// qualify, identical (author,id) pairs are deduped, and a
// permit pool caps concurrent fetch tasks. The chat TEXT
// is always forwarded (it's sanitized at the UI edge);
// only the fetch is bounded.
if let Some(att) = attachment.clone() {
let is_image = att.kind == crate::files::AttachmentKind::Image;
let key = (from, att.id);
let already_inflight =
inflight_attachments.lock().unwrap().contains(&key);
if should_auto_fetch(
is_image,
// Membership was proven by the roster name
// gate above, which drops non-members before
// any attachment handling.
true,
already_inflight,
att.size,
) && auto_fetch_budget.admit(from, att.size, unix_now_ms())
{
// Reserve the dedup slot, then a permit. If the
// pool is exhausted, drop the auto-fetch (and the
// dedup marker) — the descriptor still shows and
// the user can fetch on demand.
inflight_attachments.lock().unwrap().insert(key);
match attachment_limiter.clone().try_acquire_owned() {
Ok(permit) => {
spawn_attachment_fetch(
transport_events.clone(),
ui_tx_events.clone(),
from,
att,
true,
Some(AutoFetchGuard {
_permit: permit,
inflight: inflight_attachments.clone(),
key,
}),
);
}
Err(_) => {
inflight_attachments.lock().unwrap().remove(&key);
crate::log_msg(
"Chat attachment auto-fetch limit reached; skipping (fetch on demand)",
);
}
}
}
}
let _ = ui_tx_events
.send(UiEvent::ChatMessage {
from: from.to_string(),
name,
text,
attachment,
})
.await;
}
RoomEvent::ClockSkewSuspected { author, skew_ms } => {
crate::log_msg(&format!(
"Clock skew suspected for authenticated gossip author={} skew_ms={skew_ms}",
crate::short_id(&author.to_string())
));
let skew_secs = skew_ms.unsigned_abs().saturating_add(999) / 1000;
let _ = ui_tx_events
.send(UiEvent::ClockSkewWarning {
skew_secs,
peer_ahead: skew_ms > 0,
})
.await;
}
RoomEvent::PeerConnectionLost(peer_id) => {
// Transient drop: do NOT tear down the peer. Its audio
// supervisor stays alive and keeps redialing the
// retained address, so show "reconnecting" and arm a
// grace timer that evicts the peer only if the link
// hasn't recovered within RECONNECT_GRACE. A gossip
// rejoin (PeerJoined/PeerUpdated) or a transport
// reconnect (ConnEvent::Connected) cancels it first.
transport_events.keep_audio_sender_for_reconnect_grace(peer_id);
let _ = ui_tx_events
.send(UiEvent::PeerConnecting { id: peer_id })
.await;
arm_grace_timer(
&grace_timers_events,
&seen_connected_events,
GraceExpiry {
transport: &transport_events,
jitter: &jitter_events,
ui_tx: &ui_tx_events,
recovery: Some(&recovery_events),
chat_roster: &roster,
},
RECONNECT_GRACE,
peer_id,
);
}
}
}
});
// 5. Connection-state forwarder: turns transport link state into
// per-peer UI indicators (connecting / reconnecting vs. live).
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;
}
};
let conn_handler = ConnEventHandler::new(
ui_tx.clone(),
grace_timers.clone(),
seen_connected.clone(),
transport.clone(),
jitter.clone(),
)
.with_recovery(recovery_context)
.with_chat_roster(chat_roster.clone());
let conn_event_task = tokio::spawn(async move {
while let Some(event) = conn_events.recv().await {
conn_handler.handle(event).await;
}
});
// Connection-transparency poll: ~1/sec, snapshot every live audio
// link's selected path and hand the UI derived badge info (path
// type, RTT, window loss/bitrate). Read-only against the
// transport; owns the previous-snapshot map the derivation diffs
// against.
let transport_stats = transport.clone();
let ui_tx_stats = ui_tx.clone();
let conn_stats_task = tokio::spawn(async move {
let mut prev: HashMap<EndpointId, crate::network::PathSnapshot> =
HashMap::new();
let mut last = tokio::time::Instant::now();
let mut ticker = tokio::time::interval(connstats::POLL_INTERVAL);
ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
loop {
ticker.tick().await;
let now = tokio::time::Instant::now();
let elapsed = now - last;
last = now;
let snaps = transport_stats.connection_stats();
let infos = snaps
.iter()
.map(|(id, cur)| (*id, connstats::derive(prev.get(id), cur, elapsed)))
.collect();
prev = snaps.into_iter().collect();
if ui_tx_stats
.send(UiEvent::ConnectionStats(infos))
.await
.is_err()
{
break;
}
}
});
let session = ActiveSession {
room_state: room_state.clone(),
capture_thread,
datagram_task,
mixer_task,
event_task,
conn_event_task,
conn_stats_task,
recovery_task,
recovery_terminal_task,
grace_timers,
transport: transport.clone(),
#[cfg(target_os = "linux")]
teardown: SessionTeardown::new(echo_cancel_guard),
#[cfg(not(target_os = "linux"))]
teardown: SessionTeardown::new(None),
};
let self_id = endpoint.id().to_string();
// Member-issued ticket (W7 P3): the ticket we hand to the UI for
// sharing is stamped with OUR OWN live address + the room's topic,
// not the (possibly someone else's) ticket we joined with. So every
// member — not just the creator — hands out a working door pointing
// at themselves, which is what lets a room outlive its creator. We
// keep `ticket_str` untouched for joining + the A8 retain logic; we
// only re-stamp the *display* copy. Re-stamping is a no-op for the
// creator (same addr+topic), and a no-op if the ticket can't be
// parsed (a malformed join, which fails anyway).
let share_ticket = PeerSpeakTicket::restamp(&ticket_str, endpoint.addr());
// Advertise this gathering to friends who ping us (W7 B2): our own
// restamped member ticket → a one-click Join. The cosmetic label
// rides in the ticket (set by the creator), so every member —
// creator or joiner — reports the same room name. Re-sanitize the
// parsed label since the ticket is peer-supplied (untrusted).
let room_label =
crate::sanitize::sanitize_name(&PeerSpeakTicket::label_of(&share_ticket));
*current_room.lock().unwrap() = Some(crate::presence::RoomPresence {
name: room_label,
ticket: share_ticket.clone(),
});
let _ = ui_tx
.send(UiEvent::RoomJoined {
ticket: share_ticket,
self_id,
})
.await;
active_session = Some(session);
}
CoreCommand::Leave => {
// Finalize any recording first, while the audio feeders are alive.
stop_recording(
&recorder,
&is_recording,
&multitrack,
&is_multitrack,
&ui_tx,
)
.await;
current_sharing = None;
if let Some(session) = active_session.take() {
session.shutdown(audio_backend.clone()).await;
// Stop routing inbound audio links — the endpoint/router stay up.
net.audio_router.clear();
net.file_router.clear();
// No longer in a gathering — friends who ping see us as just online.
*current_room.lock().unwrap() = None;
let _ = ui_tx.send(UiEvent::RoomLeft).await;
}
// Apply any network-mode / identity change that was deferred while we
// were in the call (rebuild while idle keeps the endpoint reachable).
if net_rebuild_pending {
let publish = presence_mode.lock().unwrap().publishes_to_discovery();
let (stack, live) = replace_net_stack(
net,
"Applying deferred network settings",
&secret_key,
network_mode,
net_mode,
&friends_handler,
publish,
&ui_tx,
)
.await?;
net = stack;
net_mode = live;
network_mode = live;
net_rebuild_pending = false;
}
}
CoreCommand::ToggleMute => {
let current = is_muted.load(Ordering::Relaxed);
let new_state = !current;
is_muted.store(new_state, Ordering::Relaxed);
if let Some(session) = &active_session {
let self_state = presence.to_state(
new_state,
net.endpoint.addr(),
current_sharing.as_ref().map(|s| s.ticket.clone()),
);
let _ = session.room_state.update_self_state(self_state).await;
}
}
CoreCommand::SetAvatar(avatar) => {
presence.avatar = avatar;
// Re-announce presence so the room (incl. late joiners, via the
// retained presence) picks up the new avatar (W4).
if let Some(session) = &active_session {
let self_state = presence.to_state(
is_muted.load(Ordering::Relaxed),
net.endpoint.addr(),
current_sharing.as_ref().map(|s| s.ticket.clone()),
);
let _ = session.room_state.update_self_state(self_state).await;
}
}
CoreCommand::ToggleDeafen => {
let current = is_deafened.load(Ordering::Relaxed);
is_deafened.store(!current, Ordering::Relaxed);
}
CoreCommand::SetPttMode(enabled) => {
ptt_mode.store(enabled, Ordering::Relaxed);
}
CoreCommand::SetPttActive(active) => {
ptt_active.store(active, Ordering::Relaxed);
}
CoreCommand::SetPeerVolume(peer_id, vol) => {
let mut guard = peer_volumes.lock().await;
guard.insert(peer_id, vol);
}
CoreCommand::SetPeerEq(peer_id, settings) => {
let settings = settings.clamped();
let mut guard = peer_eq.lock().await;
if settings.is_flat() {
guard.remove(&peer_id);
} else {
guard.insert(peer_id, settings);
}
}
CoreCommand::SetPeerPan(peer_id, pan) => {
let pan = pan.clamp(-1.0, 1.0);
let mut guard = peer_pan.lock().await;
if pan.abs() <= 0.001 {
guard.remove(&peer_id);
} else {
guard.insert(peer_id, pan);
}
}
CoreCommand::SetPeerGate(peer_id, threshold) => {
let threshold = threshold.clamp(0.0, 1.0);
let mut guard = peer_gate.lock().await;
if threshold <= 0.0 {
guard.remove(&peer_id);
} else {
guard.insert(peer_id, threshold);
}
}
CoreCommand::SetPeerMuted(peer_id, muted) => {
let mut guard = locally_muted.lock().await;
if muted {
guard.insert(peer_id);
} else {
guard.remove(&peer_id);
}
}
CoreCommand::SetNoiseGateThreshold(threshold) => {
noise_gate_threshold.store(threshold.to_bits(), Ordering::Relaxed);
}
CoreCommand::SetInputVolume(vol) => {
input_gain.store(vol.to_bits(), Ordering::Relaxed);
}
CoreCommand::SetOutputVolume(vol) => {
output_gain.store(vol.to_bits(), Ordering::Relaxed);
}
CoreCommand::SetMicMonitor {
enabled,
input_device,
} => {
// During a call the in-call capture thread already reports the
// mic level, and it owns the backend's capture stream — leave it be.
if active_session.is_some() {
continue;
}
if enabled {
if mic_monitor.is_none() {
let (tx, rx) = std::sync::mpsc::channel();
match audio_backend.start_capture(tx, input_device) {
Ok(()) => {
let ui = ui_tx.clone();
let gain = input_gain.clone();
let thread =
std::thread::spawn(move || run_mic_monitor(rx, ui, gain));
mic_monitor = Some(MicMonitor { thread });
}
Err(e) => {
let _ = ui_tx
.send(UiEvent::Error(format!("Mic test unavailable: {e}")))
.await;
}
}
}
} else {
stop_mic_monitor(&audio_backend, mic_monitor.take());
}
}
CoreCommand::SetNetworkMode(mode) => {
// Skip when the posture is unchanged. The GUI re-sends the saved
// network mode as part of its startup config-sync, and that mode
// usually already matches the freshly-built stack — rebuilding the
// iroh endpoint for an identical posture just churns the network
// and adds a needless ~1s teardown+rebuild bounce at every launch
// (seen on both Linux and Windows/Wine). A real change still
// rebuilds exactly as before.
if mode != network_mode {
network_mode = mode;
// Rebuild the persistent stack to the new posture immediately if
// idle; if a call is active, defer to the next Leave/Join so the
// live call isn't disrupted (preserves "applies on next join").
if active_session.is_none() {
let publish = presence_mode.lock().unwrap().publishes_to_discovery();
let (stack, live) = replace_net_stack(
net,
"Network mode change",
&secret_key,
network_mode,
net_mode,
&friends_handler,
publish,
&ui_tx,
)
.await?;
net = stack;
net_mode = live;
network_mode = live;
} else {
net_rebuild_pending = true;
}
}
}
CoreCommand::SetAudioProfile(profile) => {
// Publish the new profile to the capture thread (W12). It picks up
// the change on its next frame and re-tunes the live encoder; a
// call that starts later reads the same atomic at encoder creation.
audio_profile.store(profile.as_u8(), Ordering::Relaxed);
}
CoreCommand::RegenerateIdentity => {
// Mint + persist a fresh identity, discarding the old one. The
// persistent endpoint is rebuilt with the new key (now if idle, else
// on the next Leave/Join) — consistent with "applies on next join."
let mut regenerated = false;
match crate::identity::regenerate() {
Ok(key) => {
secret_key = key;
identity_error = None;
regenerated = true;
crate::log_msg("identity: regenerated to a fresh persistent id");
}
Err(e) => {
// Couldn't write the new key — keep the current one in
// memory but report the disk problem as a degraded state.
identity_error = Some(format!("{e:#}"));
crate::log_msg(&format!("identity: regenerate failed: {e:#}"));
}
}
// Re-bind the endpoint under the new id so our node id actually
// changes. Only on a successful regenerate (a failed one left the
// key unchanged, so a rebuild would be pointless churn).
if regenerated {
if active_session.is_none() {
let publish = presence_mode.lock().unwrap().publishes_to_discovery();
// Same mode both attempts — the fallback is a plain
// retry under the (already persisted) new key.
let (stack, live) = replace_net_stack(
net,
"Endpoint restart after identity change",
&secret_key,
network_mode,
net_mode,
&friends_handler,
publish,
&ui_tx,
)
.await?;
net = stack;
net_mode = live;
} else {
net_rebuild_pending = true;
}
}
let _ = ui_tx
.send(UiEvent::IdentityStatus {
node_id: secret_key.public().to_string(),
persisted: identity_error.is_none(),
error: identity_error.clone(),
})
.await;
}
CoreCommand::AddFriend { id, name, addr } => {
// Idempotent: re-adding an existing id is a no-op (preserves the
// local name/addr), so only persist+emit when something changed.
let changed = friends.lock().unwrap().add(id, name, addr);
if changed {
persist_and_emit_friends(&friends, friends_read_only, &ui_tx).await;
}
}
CoreCommand::RemoveFriend(id) => {
if friends.lock().unwrap().remove(&id) {
persist_and_emit_friends(&friends, friends_read_only, &ui_tx).await;
}
}
CoreCommand::RenameFriend(id, new_name) => {
if friends.lock().unwrap().rename(&id, new_name) {
persist_and_emit_friends(&friends, friends_read_only, &ui_tx).await;
}
}
CoreCommand::RefreshFriends => {
// Manual "Rescan": run an immediate probe pass (same as a scheduler
// tick), detached so it can't block command handling. Honour
// Invisible — stay fully dark and touch no friend's machine. A
// `FriendsRescanned` event always follows so the UI's transient
// "Rescanning…" status clears even when probing was skipped.
let visible =
*presence_mode.lock().unwrap() != crate::presence::PresenceMode::Invisible;
let endpoint = net.endpoint.clone();
let snapshot = friends.lock().unwrap().clone();
let tx = ui_tx.clone();
tokio::spawn(async move {
if visible {
probe_friends_once(endpoint, snapshot, tx.clone()).await;
}
let _ = tx.send(UiEvent::FriendsRescanned).await;
});
}
CoreCommand::SetPresenceMode(mode) => {
let previous_mode = *presence_mode.lock().unwrap();
let now = tokio::time::Instant::now();
if previous_mode == mode {
// Same-mode requests are no-ops for discovery wiring, but keep the
// existing UX: re-selecting Discoverable restarts the clock.
discovery_deadline = if mode == PresenceMode::Discoverable {
Some(now + crate::discovery::DISCOVERY_TIMEBOX)
} else {
None
};
continue;
}
// W7 P6/S11: re-apply n0 DNS discovery for the requested posture
// first, then commit the presence mode only if the endpoint accepted
// that discovery plan. This keeps the UI truthful when dropping the
// publisher fails.
let plan =
crate::discovery::lookup_plan(network_mode, mode.publishes_to_discovery());
let apply_result = apply_discovery(&net.endpoint, &net.memory_lookup, plan);
let (committed_mode, transition_error) =
crate::discovery::resolve_presence_transition(
previous_mode,
mode,
apply_result.is_ok(),
);
*presence_mode.lock().unwrap() = committed_mode;
if committed_mode == PresenceMode::Discoverable {
if apply_result.is_ok() && mode == PresenceMode::Discoverable {
discovery_deadline = Some(now + crate::discovery::DISCOVERY_TIMEBOX);
} else {
arm_discovery_retry(&mut discovery_deadline, now);
}
} else {
discovery_deadline = None;
}
if let Err(e) = apply_result {
crate::log_msg(&format!("discovery: apply failed: {e:#}"));
if committed_mode != mode {
let _ = ui_tx
.send(UiEvent::PresenceModeReverted {
mode: committed_mode,
})
.await;
}
if let Some(message) = transition_error {
let _ = ui_tx
.send(UiEvent::Error(format!("{message} ({e:#})")))
.await;
}
}
}
CoreCommand::SetGamePresenceEnabled(enabled) => {
game_presence_enabled = enabled;
// Recompute our broadcast label: the current game when enabling,
// cleared when disabling. Publish immediately (D8) so peers see the
// game appear/disappear at once, not on the next detector tick.
presence.game = if enabled {
game_presence_label(current_game.as_ref())
} else {
None
};
if let Some(session) = &active_session {
let self_state = presence.to_state(
is_muted.load(Ordering::Relaxed),
net.endpoint.addr(),
current_sharing.as_ref().map(|s| s.ticket.clone()),
);
let _ = session.room_state.update_self_state(self_state).await;
}
}
CoreCommand::SetGameOverride(override_) => {
// Applied on the detector's next poll, immediately (bypasses debounce).
if let Some(detector) = &game_detector {
detector.set_override(override_);
}
}
CoreCommand::SetGameProcessMap(map) => {
if let Some(detector) = &game_detector {
detector.set_process_map(map);
}
}
CoreCommand::SetRecordingMode(mode) => {
recording_mode = mode;
}
CoreCommand::SetRecording(enabled) => {
if enabled {
// Only record while in a call, and not already recording.
if active_session.is_none() {
let _ = ui_tx
.send(UiEvent::Error("Join a call before recording".into()))
.await;
} else if !is_recording.load(Ordering::Relaxed) {
match dirs::home_dir() {
Some(home) => {
let base = home.join("peerspeak-recordings");
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0);
let result: Result<String, String> = if recording_mode
.is_multitrack()
{
// Multitrack/Both: a per-session directory of stems.
std::fs::create_dir_all(&base)
.map_err(|e| e.to_string())
.and_then(|_| {
crate::audio::multitrack::create_session_dir(&base, now)
.map_err(|e| e.to_string())
})
.and_then(|session_dir| {
MultitrackRecorder::create(
&session_dir,
FRAME_SAMPLES,
matches!(recording_mode, RecordingMode::Both),
)
.map_err(|e| e.to_string())
})
.map(|mut mt| {
// Register everyone already in the room so their
// stems are silence-aligned from t=0.
if let Some(session) = &active_session {
for (id, st) in session.room_state.active_peers() {
if let Err(e) = mt.add_peer(id, &st.name) {
crate::log_msg(&format!(
"multitrack add_peer failed: {e}"
));
}
}
}
let path = mt.dir().to_string_lossy().to_string();
*multitrack.lock().unwrap() = Some(mt);
is_multitrack.store(true, Ordering::Relaxed);
path
})
} else {
// Mixed: one file (original behaviour).
std::fs::create_dir_all(&base)
.map_err(|e| e.to_string())
.and_then(|_| {
crate::audio::recorder::Recorder::create(&base, now)
.map_err(|e| e.to_string())
})
.map(|rec| {
let path = rec.path().to_string_lossy().to_string();
*recorder.lock().unwrap() = Some(rec);
path
})
};
match result {
Ok(path) => {
is_recording.store(true, Ordering::Relaxed);
crate::log_msg(&format!(
"Recording started ({recording_mode:?}): {path}"
));
let _ =
ui_tx.send(UiEvent::RecordingStarted { path }).await;
}
Err(e) => {
let _ = ui_tx
.send(UiEvent::Error(format!("Recording failed: {e}")))
.await;
}
}
}
None => {
let _ = ui_tx
.send(UiEvent::Error("No home directory for recordings".into()))
.await;
}
}
}
} else {
stop_recording(
&recorder,
&is_recording,
&multitrack,
&is_multitrack,
&ui_tx,
)
.await;
}
}
CoreCommand::SendChat { local_id, text } => {
// Honest result either way (Phase 5): no active session is a
// FAILURE the sender must see, not a silent drop.
let error = match &active_session {
Some(session) => session
.room_state
.send_chat(text, None)
.await
.err()
.map(|e| e.to_string()),
None => Some("not in a room".to_string()),
};
if let Some(e) = &error {
crate::log_msg(&format!("Failed to send chat: {e}"));
}
let _ = ui_tx
.send(UiEvent::ChatSendResult { local_id, error })
.await;
}
CoreCommand::SendChatFile {
local_id,
text,
attachment,
data,
} => {
let error = match &active_session {
Some(session) => {
// Make the bytes fetchable by room members, then broadcast
// the descriptor alongside the (possibly empty) caption
// text. Re-serving the same id on a retry REPLACES the
// store entry (same Arc), never double-counts it.
session.transport.serve_attachment(attachment.id, data);
session
.room_state
.send_chat(text, Some(attachment))
.await
.err()
.map(|e| e.to_string())
}
None => Some("not in a room".to_string()),
};
if let Some(e) = &error {
crate::log_msg(&format!("Failed to send chat file: {e}"));
}
let _ = ui_tx
.send(UiEvent::ChatSendResult { local_id, error })
.await;
}
CoreCommand::FetchAttachment { from, attachment } => {
if let Some(session) = &active_session {
let is_image = attachment.kind == crate::files::AttachmentKind::Image;
// User-initiated (the "Save" chip): not bounded here — a human
// click rate-limits it. The auto path (F-02) passes a guard.
spawn_attachment_fetch(
session.transport.clone(),
ui_tx.clone(),
from,
attachment,
is_image,
None,
);
}
}
CoreCommand::ServeMusicTrack { id, data } => {
if let Some(session) = &active_session {
session.transport.serve_attachment(id, data);
}
}
CoreCommand::ForgetMusicTrack(id) => {
if let Some(session) = &active_session {
session.transport.forget_attachment(id);
}
}
CoreCommand::SetMusicPresence(music) => {
presence.music = music;
if let Some(session) = &active_session {
let self_state = presence.to_state(
is_muted.load(Ordering::Relaxed),
net.endpoint.addr(),
current_sharing.as_ref().map(|s| s.ticket.clone()),
);
let _ = session.room_state.update_self_state(self_state).await;
}
}
CoreCommand::FetchMusic { from, id, size } => {
if let Some(session) = &active_session {
spawn_music_fetch(session.transport.clone(), ui_tx.clone(), from, id, size);
}
}
CoreCommand::PrefetchMusic { from, id, size } => {
if let Some(session) = &active_session {
spawn_music_prefetch(session.transport.clone(), ui_tx.clone(), from, id, size);
}
}
CoreCommand::SetPixelpassPath(path) => {
pixelpass_override = path.filter(|p| !p.trim().is_empty());
}
CoreCommand::ListAudioApps => {
// Probe whether this pixelpass supports `--strict-audio` before
// offering per-app capture: an older binary would reject the flag
// and hard-fail the share (audit P2). When unsupported (or
// pixelpass is missing), skip enumeration and let the picker show
// whole-desktop audio only — never a best-effort `--app` that
// would reopen the A23 echo.
let app_audio_supported =
match crate::screenshare::pixelpass_path(pixelpass_override.as_deref()) {
Some(bin) => crate::screenshare::supports_strict_audio(&bin).await,
None => false,
};
let apps = if app_audio_supported {
crate::screenshare::list_audio_apps().await
} else {
Vec::new()
};
let _ = ui_tx
.send(UiEvent::AudioAppsListed {
apps,
app_audio_supported,
})
.await;
}
CoreCommand::StartScreenShare {
audio_app,
settings,
quality,
} => {
let Some(session) = &mut active_session else {
let _ = ui_tx
.send(UiEvent::Error(
"Join a call before sharing your screen".into(),
))
.await;
continue;
};
if session.teardown.is_sharing() {
continue; // already sharing
}
let bin = match crate::screenshare::pixelpass_path(pixelpass_override.as_deref()) {
Some(b) => b,
None => {
let _ = ui_tx
.send(UiEvent::Error(
"pixelpass not found — install it to share your screen".into(),
))
.await;
continue;
}
};
// Every share gets a notice forwarder — not just app-audio ones.
// pixelpass `app_audio` events (only emitted when an app is
// selected) become UI warnings, and the drain's terminal `Eof`
// becomes a host fault scoped to this spawn's generation, so a
// host that dies is torn down instead of staying advertised in
// presence forever. On a failed spawn the sender is dropped
// before the drain ever runs, so the forwarder just ends and no
// fault is sent (the spawn error carries the news instead).
share_generations += 1;
let generation = share_generations;
let (notices_tx, mut notices_rx) =
tokio::sync::mpsc::unbounded_channel::<crate::screenshare::HostNotice>();
let ui_tx_notices = ui_tx.clone();
let fault_tx = host_fault_tx.clone();
tokio::spawn(async move {
while let Some(notice) = notices_rx.recv().await {
match notice {
crate::screenshare::HostNotice::Event(ev) => {
let active = match ev {
crate::screenshare::PixelpassEvent::AppAudioRouted => true,
crate::screenshare::PixelpassEvent::AppAudioLost => false,
_ => continue,
};
if ui_tx_notices
.send(UiEvent::ShareAudioActive(active))
.await
.is_err()
{
break;
}
}
// Terminal by contract: nothing follows on the
// channel, so the task ends here.
crate::screenshare::HostNotice::Eof => {
let _ = fault_tx.send(generation);
break;
}
}
}
});
match crate::screenshare::spawn_host(
&bin,
audio_app.as_deref(),
&settings,
quality,
notices_tx,
)
.await
{
Ok((child, ticket)) => {
crate::log_msg("Screen share host started");
session.teardown.set_host(child);
current_sharing = Some(ActiveShare {
generation,
ticket: ticket.clone(),
});
let self_state = presence.to_state(
is_muted.load(Ordering::Relaxed),
net.endpoint.addr(),
Some(ticket),
);
let _ = session.room_state.update_self_state(self_state).await;
let _ = ui_tx.send(UiEvent::ScreenShareStarted).await;
}
Err(e) => {
let _ = ui_tx
.send(UiEvent::Error(format!("Couldn't start screen share: {e}")))
.await;
}
}
}
CoreCommand::StopScreenShare => {
current_sharing = None;
if let Some(session) = &mut active_session {
match session.teardown.stop_host().await {
None => {}
Some(teardown::StopOutcome::Reaped) => {
crate::log_msg("Screen share host stopped");
}
// We gave up waiting rather than freeze the client, so
// pixelpass may still be alive and serving. Saying
// "stopped" and nothing else would be a lie the user
// cannot see through (round-16 review, P3-2).
Some(teardown::StopOutcome::Unconfirmed) => {
let _ = ui_tx
.send(UiEvent::Error(
"Couldn't confirm the screen-share process exited — \
it may still be sharing. Check for a stray pixelpass."
.into(),
))
.await;
}
}
let self_state = presence.to_state(
is_muted.load(Ordering::Relaxed),
net.endpoint.addr(),
None,
);
let _ = session.room_state.update_self_state(self_state).await;
}
let _ = ui_tx.send(UiEvent::ScreenShareStopped).await;
}
CoreCommand::ScreenShareHostFault { generation } => {
// Stale unless it names the share we are advertising RIGHT NOW.
// Every deliberate end of a share (StopScreenShare, Leave, a
// fresh Join) clears `current_sharing` before or while reaping
// the child, and the reaped child's stdout EOF then arrives
// here late — dropping it is the correct handling, not an edge
// case. A mismatched generation likewise: that fault belongs to
// an older spawn than the share now running.
let stale = current_sharing.as_ref().map(|s| s.generation) != Some(generation);
if stale {
continue;
}
crate::log_msg(
"Screen share host died (stdout EOF with the share still advertised)",
);
current_sharing = None;
// Pull the ticket off presence FIRST, before the reap: if the
// child only closed stdout and lives on, `stop_host` burns the
// full stop grace before the SIGKILL fallback, and for that
// whole window peers would still see (and click Watch on) a
// share whose host is already gone (Gemini S2-merge review,
// P2-1).
if let Some(session) = &mut active_session {
let self_state = presence.to_state(
is_muted.load(Ordering::Relaxed),
net.endpoint.addr(),
None,
);
let _ = session.room_state.update_self_state(self_state).await;
}
// Stopped next — it clears the UI's sharing state — so the
// local UI also stops saying "sharing" before the reap wait,
// and the error explaining why comes only after, so the user
// is never left looking at a "sharing" UI with an error
// beside it.
let _ = ui_tx.send(UiEvent::ScreenShareStopped).await;
let mut unconfirmed = false;
if let Some(session) = &mut active_session {
// The child is usually already dead, so this confirms the
// reap immediately; if it merely closed stdout and lives
// on, this is the SIGINT → grace → SIGKILL path. Either
// way the dead-or-dying child leaves the teardown slot, so
// `is_sharing` stops lying.
unconfirmed = matches!(
session.teardown.stop_host().await,
Some(teardown::StopOutcome::Unconfirmed)
);
}
let detail = if unconfirmed {
" Its process also couldn't be confirmed dead — check for a stray pixelpass."
} else {
""
};
let _ = ui_tx
.send(UiEvent::Error(format!(
"Screen share ended unexpectedly — pixelpass exited.{detail}"
)))
.await;
}
CoreCommand::ViewShare { ticket, settings } => {
let bin = match crate::screenshare::pixelpass_path(pixelpass_override.as_deref()) {
Some(b) => b,
None => {
let _ = ui_tx
.send(UiEvent::Error(
"pixelpass not found — install it to watch screen shares".into(),
))
.await;
continue;
}
};
if let Some(session) = &mut active_session {
// Drop viewers whose player window has already closed so the
// list only tracks live players.
session.teardown.sweep_exited_viewers();
// One player per share: a second Watch click on a share we're
// already viewing is a retry (usually because the first window
// froze), so replace the existing player rather than stacking a
// second mpv — two players would double the shared audio.
if session.teardown.replace_viewer(&ticket).await {
crate::log_msg("Screen share viewer replaced (re-watch)");
}
}
match crate::screenshare::spawn_viewer(&bin, &ticket, &settings).await {
Ok(child) => {
crate::log_msg("Screen share viewer started");
if let Some(session) = &mut active_session {
session.teardown.push_viewer(ticket, child);
}
}
Err(e) => {
let _ = ui_tx
.send(UiEvent::Error(format!("Couldn't watch screen share: {e}")))
.await;
}
}
}
}
}
// The command loop has exited, by any route. Tear the session down
// explicitly rather than letting it drop on the way out of this function:
// an implicit drop unloads the echo-cancel module without first reaping the
// pixelpass host (design v3.4 §7.2, decision D4).
//
// This sits *after* the loop rather than in the close arm on purpose. The
// impl plan pinned one teardown per channel-close arm, but the best-effort
// wake arm is unreachable by construction (see the comment at that arm), so
// that shape would have duplicated teardown to cover one live path and one
// dead one. Here every `break` is covered structurally, including any added
// later. Adjudication: impl plan §10, 2026-07-26.
if let Some(session) = active_session.take() {
session.shutdown(audio_backend.clone()).await;
}
Ok(())
}
/// Index of an existing viewer for `ticket` in the live-viewers list, if any.
/// A re-watch of the same share replaces that player instead of stacking a
/// second one — two players decoding the same stream would double the shared
/// audio. Generic over the child value so the dedup rule is unit-testable
/// without spawning real player processes.
fn replace_viewer_index<T>(viewers: &[(String, T)], ticket: &str) -> Option<usize> {
viewers.iter().position(|(t, _)| t == ticket)
}
#[cfg(test)]
mod tests {
use super::{
KnownPeers, MAX_OPUS_PAYLOAD, MAX_RETAINED_PEERS, MIC_LEVEL_REPORT_SAMPLES, MicLevelMeter,
NetworkMode, PLAYBACK_HANDOFF_QUEUE_FRAMES, PeerSpeakTicket, admit_retained,
apply_peer_volume, apply_volume, audio_datagram_len_ok, coalesce_insert, coalesce_pop,
frame_level, mix_frames, mix_stereo_frames, next_game_change, rebuild_with_fallback,
replace_viewer_index, send_playback_frame, should_auto_fetch, stereo_to_mono,
};
use crate::core::messages::{CoalesceKey, CoreCommand, coalesce_key};
use std::collections::{HashMap, HashSet};
use std::sync::mpsc::sync_channel;
use std::time::Duration;
fn endpoint_id() -> iroh::EndpointId {
iroh::SecretKey::generate().public()
}
#[test]
fn re_watch_replaces_existing_viewer_for_same_ticket() {
// The value type stands in for a viewer Child; only the ticket matters.
let viewers = vec![("ticket-A".to_string(), 0u8), ("ticket-B".to_string(), 1u8)];
// Re-watching an already-open share finds the existing player to replace.
assert_eq!(replace_viewer_index(&viewers, "ticket-A"), Some(0));
assert_eq!(replace_viewer_index(&viewers, "ticket-B"), Some(1));
// A different (new) share has nothing to replace — it opens fresh.
assert_eq!(replace_viewer_index(&viewers, "ticket-C"), None);
// Empty list: first watch of anything opens fresh.
assert_eq!(replace_viewer_index::<u8>(&[], "ticket-A"), None);
}
// --- rebuild_with_fallback: the retry policy behind replace_net_stack ---
// The builder is injected, so these cover the policy without sockets. The
// closure does its bookkeeping synchronously and returns a ready future.
#[tokio::test]
async fn rebuild_keeps_requested_posture_on_first_success() {
let calls = std::cell::RefCell::new(Vec::new());
let out = rebuild_with_fallback(
|mode| {
calls.borrow_mut().push(mode);
std::future::ready(Ok::<u8, String>(7))
},
NetworkMode::DirectOnly,
NetworkMode::N0Full,
)
.await;
assert_eq!(out, Ok((7, NetworkMode::DirectOnly, None)));
// No second build: the live posture is only a fallback.
assert_eq!(*calls.borrow(), vec![NetworkMode::DirectOnly]);
}
#[tokio::test]
async fn rebuild_falls_back_to_the_live_posture_when_the_requested_one_fails() {
let calls = std::cell::RefCell::new(Vec::new());
let out = rebuild_with_fallback(
|mode| {
calls.borrow_mut().push(mode);
std::future::ready(if mode == NetworkMode::DirectOnly {
Err("bind failed".to_string())
} else {
Ok(7u8)
})
},
NetworkMode::DirectOnly,
NetworkMode::N0Full,
)
.await;
// A stack is up on the OLD posture and the caller learns both that it
// fell back (mode) and why (the primary error) — no silent zombie.
assert_eq!(
out,
Ok((7, NetworkMode::N0Full, Some("bind failed".to_string())))
);
assert_eq!(
*calls.borrow(),
vec![NetworkMode::DirectOnly, NetworkMode::N0Full]
);
}
#[tokio::test]
async fn rebuild_reports_both_errors_when_networking_is_gone() {
let out = rebuild_with_fallback(
|_| std::future::ready(Err::<u8, String>("bind failed".to_string())),
NetworkMode::DirectOnly,
NetworkMode::N0Full,
)
.await;
assert_eq!(
out,
Err(("bind failed".to_string(), "bind failed".to_string()))
);
}
#[tokio::test]
async fn rebuild_with_equal_postures_is_a_plain_retry() {
// RegenerateIdentity rebuilds under the same mode: the fallback is a
// second attempt with identical parameters, not a posture change.
let calls = std::cell::Cell::new(0u8);
let out = rebuild_with_fallback(
|mode| {
calls.set(calls.get() + 1);
assert_eq!(mode, NetworkMode::RelayNoDiscovery);
std::future::ready(if calls.get() == 1 {
Err("transient".to_string())
} else {
Ok(7u8)
})
},
NetworkMode::RelayNoDiscovery,
NetworkMode::RelayNoDiscovery,
)
.await;
// Succeeded on the requested posture, so the caller treats the change
// as applied (the Some(err) is logged, not surfaced as a UI error).
assert_eq!(
out,
Ok((
7,
NetworkMode::RelayNoDiscovery,
Some("transient".to_string())
))
);
assert_eq!(calls.get(), 2);
}
#[test]
fn admit_retained_rejects_only_new_ids_at_the_cap() {
// Below the cap, a brand-new identity is retained.
assert!(admit_retained(0, true, MAX_RETAINED_PEERS));
assert!(admit_retained(
MAX_RETAINED_PEERS - 1,
true,
MAX_RETAINED_PEERS
));
// At the cap, a brand-new identity is refused — this is the bound that stops
// an insider grace-cycling distinct identities from growing the retain table.
assert!(!admit_retained(
MAX_RETAINED_PEERS,
true,
MAX_RETAINED_PEERS
));
// A peer already tracked always refreshes, even at (or past) the cap: it only
// updates an existing address and never adds a slot.
assert!(admit_retained(
MAX_RETAINED_PEERS,
false,
MAX_RETAINED_PEERS
));
assert!(admit_retained(
MAX_RETAINED_PEERS + 5,
false,
MAX_RETAINED_PEERS
));
}
#[test]
fn coalesce_insert_overwrites_same_peer_control_with_latest_value() {
let peer = endpoint_id();
let mut map = HashMap::new();
assert!(coalesce_insert(
&mut map,
CoreCommand::SetPeerVolume(peer, 0.2)
));
assert!(coalesce_insert(
&mut map,
CoreCommand::SetPeerVolume(peer, 0.9)
));
assert_eq!(map.len(), 1);
match coalesce_pop(&mut map) {
Some(CoreCommand::SetPeerVolume(id, volume)) => {
assert_eq!(id, peer);
assert_eq!(volume, 0.9);
}
other => panic!("expected latest peer volume, got {other:?}"),
}
assert!(map.is_empty());
}
#[test]
fn coalesce_insert_keeps_distinct_peers_for_same_control() {
let peer_a = endpoint_id();
let peer_b = endpoint_id();
let mut map = HashMap::new();
assert!(coalesce_insert(
&mut map,
CoreCommand::SetPeerVolume(peer_a, 0.2)
));
assert!(coalesce_insert(
&mut map,
CoreCommand::SetPeerVolume(peer_b, 0.9)
));
assert_eq!(map.len(), 2);
let mut seen = HashMap::new();
while let Some(cmd) = coalesce_pop(&mut map) {
match cmd {
CoreCommand::SetPeerVolume(id, volume) => {
seen.insert(id, volume);
}
other => panic!("expected peer volume command, got {other:?}"),
}
}
assert_eq!(seen.get(&peer_a), Some(&0.2));
assert_eq!(seen.get(&peer_b), Some(&0.9));
}
#[test]
fn coalesce_insert_overwrites_global_control_with_latest_value() {
let mut map = HashMap::new();
assert!(coalesce_insert(&mut map, CoreCommand::SetInputVolume(0.1)));
assert!(coalesce_insert(&mut map, CoreCommand::SetInputVolume(0.4)));
assert_eq!(map.len(), 1);
match coalesce_pop(&mut map) {
Some(CoreCommand::SetInputVolume(volume)) => assert_eq!(volume, 0.4),
other => panic!("expected latest input volume, got {other:?}"),
}
assert!(map.is_empty());
}
#[test]
fn coalesce_pop_empty_map_returns_none() {
let mut map = HashMap::new();
assert!(coalesce_pop(&mut map).is_none());
}
#[test]
fn coalesce_pop_drains_each_distinct_key_once() {
let peer_a = endpoint_id();
let peer_b = endpoint_id();
let mut map = HashMap::new();
assert!(coalesce_insert(
&mut map,
CoreCommand::SetPeerPan(peer_a, -0.3)
));
assert!(coalesce_insert(
&mut map,
CoreCommand::SetPeerGate(peer_b, 0.4)
));
assert!(coalesce_insert(&mut map, CoreCommand::SetOutputVolume(0.8)));
let expected = HashSet::from([
CoalesceKey::PeerPan(peer_a),
CoalesceKey::PeerGate(peer_b),
CoalesceKey::OutputVolume,
]);
let mut seen = HashSet::new();
while let Some(cmd) = coalesce_pop(&mut map) {
seen.insert(coalesce_key(&cmd).expect("coalesced command should have a key"));
}
assert_eq!(seen, expected);
assert!(map.is_empty());
}
#[test]
fn auto_fetch_only_for_roster_images_not_already_inflight() {
const OK_SIZE: u64 = 1024;
// The happy path: a roster author's brand-new, small-enough image.
assert!(should_auto_fetch(true, true, false, OK_SIZE));
// A non-image (generic file) never auto-fetches — it waits for "Save".
assert!(!should_auto_fetch(false, true, false, OK_SIZE));
// A non-roster author (e.g. a sock puppet that never announced) is rejected,
// closing the F-02 unbounded-task vector.
assert!(!should_auto_fetch(true, false, false, OK_SIZE));
// An identical (author,id) already being fetched is deduped.
assert!(!should_auto_fetch(true, true, true, OK_SIZE));
// The declared-size gate (Phase 3B): at the cap auto-fetches, the first
// byte over requires a click.
assert!(should_auto_fetch(
true,
true,
false,
crate::files::MAX_AUTO_IMAGE_BYTES
));
assert!(!should_auto_fetch(
true,
true,
false,
crate::files::MAX_AUTO_IMAGE_BYTES + 1
));
}
#[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]
}
#[tokio::test]
async fn closed_game_watch_is_fused_after_one_ready_event() {
let (tx, rx) = tokio::sync::watch::channel(None);
let mut rx = Some(rx);
drop(tx);
assert_eq!(next_game_change(&mut rx).await, None);
assert!(
rx.is_none(),
"closed receiver must disable its select source"
);
}
#[test]
fn mic_meter_reports_only_after_enough_samples() {
let mut m = MicLevelMeter::new();
// One short frame well under the report window yields nothing yet.
assert_eq!(m.push(&frame(1000, 480)), None);
// A frame that crosses the window boundary triggers a report.
assert!(m.push(&frame(1000, MIC_LEVEL_REPORT_SAMPLES)).is_some());
}
#[test]
fn audio_datagram_length_gate_preserves_header_and_caps_payload() {
assert!(!audio_datagram_len_ok(0));
assert!(!audio_datagram_len_ok(3));
assert!(audio_datagram_len_ok(4));
assert!(audio_datagram_len_ok(4 + MAX_OPUS_PAYLOAD));
assert!(!audio_datagram_len_ok(5 + MAX_OPUS_PAYLOAD));
}
#[tokio::test]
async fn playback_handoff_waits_for_bounded_queue_space() {
let (tx, rx) = sync_channel::<Vec<i16>>(PLAYBACK_HANDOFF_QUEUE_FRAMES);
for n in 0..PLAYBACK_HANDOFF_QUEUE_FRAMES {
tx.try_send(vec![n as i16]).unwrap();
}
let worker = std::thread::spawn(move || {
std::thread::sleep(Duration::from_millis(20));
for n in 0..PLAYBACK_HANDOFF_QUEUE_FRAMES {
assert_eq!(rx.recv().unwrap(), vec![n as i16]);
}
assert_eq!(rx.recv().unwrap(), vec![99, 100]);
});
let sent = tokio::time::timeout(
Duration::from_secs(1),
send_playback_frame(&tx, vec![99, 100]),
)
.await
.expect("bounded handoff should unblock after the worker drains a frame");
assert!(sent);
drop(tx);
worker.join().unwrap();
}
#[test]
fn mic_meter_holds_the_peak_across_the_window() {
let mut m = MicLevelMeter::new();
let chunk = MIC_LEVEL_REPORT_SAMPLES / 4;
// Loud frame first, then quiet ones — the reported value is the loud peak.
assert_eq!(m.push(&frame(8000, chunk)), None);
assert_eq!(m.push(&frame(100, chunk)), None);
assert_eq!(m.push(&frame(100, chunk)), None);
let reported = m.push(&frame(100, chunk)).expect("window complete");
let loud = frame_level(&frame(8000, chunk));
assert!(
(reported - loud).abs() < 1e-6,
"peak should hold the loud frame"
);
}
#[test]
fn mic_meter_resets_after_reporting() {
let mut m = MicLevelMeter::new();
// Fill and report a loud window.
assert!(m.push(&frame(8000, MIC_LEVEL_REPORT_SAMPLES)).is_some());
// The next window of silence must report ~zero, not the stale loud peak.
let reported = m
.push(&frame(0, MIC_LEVEL_REPORT_SAMPLES))
.expect("second window");
assert_eq!(reported, 0.0, "peak and accumulator reset between windows");
}
#[test]
fn mic_meter_silence_reports_zero() {
let mut m = MicLevelMeter::new();
let reported = m
.push(&frame(0, MIC_LEVEL_REPORT_SAMPLES))
.expect("window complete");
assert_eq!(reported, 0.0);
}
#[test]
fn mix_of_no_peers_is_silence() {
let mixed = mix_frames(&[], 4);
assert_eq!(mixed, vec![0i32; 4]);
}
#[test]
fn single_peer_passes_through_unchanged() {
let frame = vec![100, -200, 300, -400];
let mixed = mix_frames(std::slice::from_ref(&frame), 4);
assert_eq!(mixed, vec![100i32, -200, 300, -400]);
}
#[test]
fn centered_stereo_mix_is_exact_dual_mono() {
let a = vec![100, -200, 300, -400];
let b = vec![50, 200, -100, 400];
let mixed = mix_stereo_frames(&[(a, 0.0), (b, 0.0)], 4);
assert_eq!(mixed, vec![150, 150, 0, 0, 200, 200, 0, 0]);
}
#[test]
fn hard_left_pan_only_contributes_left_channel() {
let frame = vec![100, 200];
let mixed = mix_stereo_frames(&[(frame, -1.0)], 2);
assert_eq!(mixed, vec![141, 0, 283, 0]);
}
#[test]
fn stereo_fold_down_averages_pairs() {
let mono = stereo_to_mono(&[100, 100, 200, 0, i16::MAX, i16::MAX]);
assert_eq!(mono, vec![100, 100, i16::MAX]);
}
#[test]
fn two_peers_sum_sample_by_sample() {
let a = vec![100, -200, 300, -400];
let b = vec![50, 200, -100, 400];
let mixed = mix_frames(&[a, b], 4);
assert_eq!(mixed, vec![150i32, 0, 200, 0]);
}
#[test]
fn loud_positive_mix_is_lossless_not_clamped() {
// The bus is a lossless i32 sum now — the true peak (~2x i16::MAX) is
// preserved so the limiter can ride it down. (The old mixer clamped here.)
let a = vec![30_000; 4];
let b = vec![30_000; 4];
let mixed = mix_frames(&[a, b], 4);
assert_eq!(mixed, vec![60_000i32; 4]);
}
#[test]
fn loud_negative_mix_is_lossless_not_clamped() {
let a = vec![i16::MIN; 4];
let b = vec![i16::MIN; 4];
let mixed = mix_frames(&[a, b], 4);
assert_eq!(mixed, vec![2 * i16::MIN as i32; 4]);
}
#[test]
fn shorter_peer_frame_contributes_zero_past_its_end() {
let full = vec![100, 100, 100, 100];
let short = vec![10, 20]; // only first two samples
let mixed = mix_frames(&[full, short], 4);
assert_eq!(mixed, vec![110i32, 120, 100, 100]);
}
#[test]
fn volume_unity_is_a_noop() {
let mut frame = vec![100, -200, 300, -400];
apply_volume(&mut frame, 1.0);
assert_eq!(frame, vec![100, -200, 300, -400]);
}
#[test]
fn volume_zero_mutes() {
let mut frame = vec![100, -200, 300, -400];
apply_volume(&mut frame, 0.0);
assert_eq!(frame, vec![0, 0, 0, 0]);
}
#[test]
fn volume_half_scales_samples() {
let mut frame = vec![100, -200, 300, -400];
apply_volume(&mut frame, 0.5);
// 100*0.5=50, -200*0.5=-100, 300*0.5=150, -400*0.5=-200 (exact in f32 here)
assert_eq!(frame, vec![50, -100, 150, -200]);
}
#[test]
fn volume_boost_saturates_not_wraps() {
// 20000 * 4.0 = 80000, well past i16::MAX — must clamp, not wrap.
let mut frame = vec![20_000, -20_000, 20_000, -20_000];
apply_volume(&mut frame, 4.0);
assert_eq!(frame, vec![i16::MAX, i16::MIN, i16::MAX, i16::MIN]);
}
#[test]
fn frame_level_of_silence_is_zero() {
assert_eq!(frame_level(&[0, 0, 0, 0]), 0.0);
assert_eq!(frame_level(&[]), 0.0);
}
#[test]
fn frame_level_of_full_scale_is_about_one() {
let full = vec![i16::MAX; 64];
let level = frame_level(&full);
assert!(level > 0.99 && level <= 1.0, "full-scale level was {level}");
}
#[test]
fn volume_truncates_toward_zero() {
let mut frame = vec![3, -3, 5, -5];
apply_volume(&mut frame, 0.5);
assert_eq!(frame, vec![1, -1, 2, -2]);
}
#[test]
fn volume_amplifies_without_saturating() {
let mut frame = vec![1000, -1000];
apply_volume(&mut frame, 2.0);
assert_eq!(frame, vec![2000, -2000]);
}
#[test]
fn peer_volume_map_scales_the_matching_audio_peer_frame() {
let peer = iroh::SecretKey::generate().public();
let other_peer = iroh::SecretKey::generate().public();
let volumes = std::collections::HashMap::from([(peer, 0.5), (other_peer, 2.0)]);
let mut frame = vec![100, -200, 300, -400];
apply_peer_volume(&mut frame, peer, &volumes);
assert_eq!(frame, vec![50, -100, 150, -200]);
}
#[test]
fn peer_volume_map_defaults_to_unity_when_audio_peer_key_is_unmatched() {
let ui_peer = iroh::SecretKey::generate().public();
let audio_peer = iroh::SecretKey::generate().public();
let volumes = std::collections::HashMap::from([(ui_peer, 0.5)]);
let mut frame = vec![100, -200, 300, -400];
apply_peer_volume(&mut frame, audio_peer, &volumes);
assert_eq!(frame, vec![100, -200, 300, -400]);
}
#[test]
fn three_peers_sum_without_saturation() {
let a = vec![10, 20];
let b = vec![3, 4];
let c = vec![100, -50];
let mixed = mix_frames(&[a, b, c], 2);
assert_eq!(mixed, vec![113i32, -26]);
}
#[test]
fn mix_zero_pads_output_longer_than_peer_frames() {
let a = vec![100, 200];
let mixed = mix_frames(&[a], 4);
assert_eq!(mixed, vec![100i32, 200, 0, 0]);
}
#[test]
fn frame_level_mid_range() {
let frame = vec![16384; 64];
let level = frame_level(&frame);
assert!((level - 0.5).abs() < 1e-3, "mid-range level was {level}");
}
}