diff --git a/src/core/messages.rs b/src/core/messages.rs index db7bd4f..fe42575 100644 --- a/src/core/messages.rs +++ b/src/core/messages.rs @@ -137,6 +137,16 @@ pub enum CoreCommand { /// Stop sharing our screen: kill the pixelpass host and clear the presence /// ticket. No-op when not sharing. StopScreenShare, + /// **Core-internal.** The running pixelpass host's stdout ended — the + /// process died (or its event stream broke), so the share identified by + /// `generation` is over: reap the child, pull the ticket off presence, and + /// tell the user. Synthesized by the core's own notice-forwarder task; the + /// UI never sends it. `generation` scopes the fault to one specific host + /// spawn, so a stale fault (the user already stopped, or started a new + /// share) is ignored rather than tearing down the wrong share. + ScreenShareHostFault { + generation: u64, + }, /// Watch a peer's screen share: spawn a pixelpass viewer for `ticket` and /// open it in a local player. ViewShare { @@ -271,6 +281,7 @@ pub fn delivery_class(cmd: &CoreCommand) -> DeliveryClass { quality: _, } | CoreCommand::StopScreenShare + | CoreCommand::ScreenShareHostFault { generation: _ } | CoreCommand::ViewShare { ticket: _, settings: _, @@ -363,6 +374,7 @@ pub fn coalesce_key(cmd: &CoreCommand) -> Option { quality: _, } | CoreCommand::StopScreenShare + | CoreCommand::ScreenShareHostFault { generation: _ } | CoreCommand::ViewShare { ticket: _, settings: _, diff --git a/src/core/mod.rs b/src/core/mod.rs index 6a3c78e..58dcbdd 100644 --- a/src/core/mod.rs +++ b/src/core/mod.rs @@ -1397,10 +1397,25 @@ async fn run_core_loop( // later opt-in can immediately publish whatever is currently running. let mut current_game: Option = None; let mut network_mode = NetworkMode::default(); - // Pixelpass binary override (config), and the ticket of our own active screen - // share (rides our presence so the room — incl. late joiners — can watch). + // 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 = None; - let mut current_sharing: Option = None; + struct ActiveShare { + generation: u64, + ticket: String, + } + let mut current_sharing: Option = 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::(); let mut active_session: Option = None; // Standalone capture-only mic meter, live only when no session exists. @@ -1549,6 +1564,13 @@ async fn run_core_loop( // 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 { @@ -1567,7 +1589,7 @@ async fn run_core_loop( let self_state = presence.to_state( is_muted.load(Ordering::Relaxed), net.endpoint.addr(), - current_sharing.clone(), + current_sharing.as_ref().map(|s| s.ticket.clone()), ); let _ = session.room_state.update_self_state(self_state).await; } @@ -2828,8 +2850,11 @@ async fn run_core_loop( 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.clone()); + 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; } } @@ -2842,7 +2867,7 @@ async fn run_core_loop( let self_state = presence.to_state( is_muted.load(Ordering::Relaxed), net.endpoint.addr(), - current_sharing.clone(), + current_sharing.as_ref().map(|s| s.ticket.clone()), ); let _ = session.room_state.update_self_state(self_state).await; } @@ -3155,7 +3180,7 @@ async fn run_core_loop( let self_state = presence.to_state( is_muted.load(Ordering::Relaxed), net.endpoint.addr(), - current_sharing.clone(), + current_sharing.as_ref().map(|s| s.ticket.clone()), ); let _ = session.room_state.update_self_state(self_state).await; } @@ -3361,7 +3386,7 @@ async fn run_core_loop( let self_state = presence.to_state( is_muted.load(Ordering::Relaxed), net.endpoint.addr(), - current_sharing.clone(), + current_sharing.as_ref().map(|s| s.ticket.clone()), ); let _ = session.room_state.update_self_state(self_state).await; } @@ -3435,46 +3460,62 @@ async fn run_core_loop( continue; } }; - // Forward pixelpass `app_audio` events (only emitted when an app - // is selected) to the UI so it can warn when the chosen app's - // audio drops. The channel closes when the host dies (drain hits - // EOF), ending the forwarder task on its own. - let notices = audio_app.as_deref().map(|_| { - let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::< - crate::screenshare::PixelpassEvent, - >(); - let ui_tx_notices = ui_tx.clone(); - tokio::spawn(async move { - while let Some(ev) = rx.recv().await { - 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() - { + // 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::(); + 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; } } - }); - tx + } }); match crate::screenshare::spawn_host( &bin, audio_app.as_deref(), &settings, quality, - notices, + notices_tx, ) .await { Ok((child, ticket)) => { crate::log_msg("Screen share host started"); session.teardown.set_host(child); - current_sharing = Some(ticket.clone()); + current_sharing = Some(ActiveShare { + generation, + ticket: ticket.clone(), + }); let self_state = presence.to_state( is_muted.load(Ordering::Relaxed), net.endpoint.addr(), @@ -3523,6 +3564,58 @@ async fn run_core_loop( 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; + 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) + ); + // Pull the ticket off presence so nobody clicks Watch on a + // dead share. + 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 first — it clears the UI's sharing state — and only + // then the error explaining why, so the user is never left + // looking at a "sharing" UI with an error beside it. + let _ = ui_tx.send(UiEvent::ScreenShareStopped).await; + 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, diff --git a/src/screenshare/mod.rs b/src/screenshare/mod.rs index 839a6fe..982fbd0 100644 --- a/src/screenshare/mod.rs +++ b/src/screenshare/mod.rs @@ -90,6 +90,21 @@ pub enum PixelpassEvent { Other, } +/// What the host's stdout drain forwards to the core over the notice channel. +/// +/// `Eof` is **synthesized here**, not parsed: pixelpass has no "I died" event, +/// and a crash can abort across `extern "C"` before any JSON line is written, +/// so the stream ending is the only reliable death signal. A read *error* +/// counts too — either way the event stream is gone and the host must be +/// treated as over. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum HostNotice { + /// A parsed pixelpass event line. + Event(PixelpassEvent), + /// The host's stdout ended (EOF or read error). Terminal: nothing follows. + Eof, +} + /// Parse a single stdout line from pixelpass `--output json`. Pure: no I/O. pub fn parse_pixelpass_event(line: &str) -> Option { let line = line.trim(); @@ -370,7 +385,10 @@ pub fn is_available(config_override: Option<&str>) -> bool { /// `audio_app` is `Some`, pixelpass captures only that app's audio instead of the /// whole desktop sink, which avoids the call-loopback echo (A23). The child keeps /// running (streaming to viewers) until killed or dropped; remaining stdout is -/// drained in a background task so a full pipe can't stall the host. We do +/// drained in a background task so a full pipe can't stall the host. The drain +/// forwards every parsed event over `notices` and — the part no share may opt +/// out of — a terminal [`HostNotice::Eof`] when the stream ends, which is the +/// caller's only reliable signal that the host died. We do /// not pass encode/viewer overrides unless the local settings explicitly ask for /// them, so pixelpass keeps its own defaults in the common case. pub async fn spawn_host( @@ -378,7 +396,7 @@ pub async fn spawn_host( audio_app: Option<&str>, settings: &ScreenShareSettings, quality: ShareQuality, - notices: Option>, + notices: tokio::sync::mpsc::UnboundedSender, ) -> std::io::Result<(Child, String)> { let args = host_args(audio_app, settings, quality); // Log the exact argv we hand pixelpass so a field log can confirm which @@ -433,7 +451,7 @@ pub async fn spawn_host( if let Some(stderr) = stderr { drain_stderr_in_background(stderr); } - drain_in_background(lines, "host", notices); + drain_in_background(lines, "host", Some(notices)); Ok((child, ticket)) } @@ -572,13 +590,15 @@ where /// Keep reading the child's stdout to EOF in the background so a full pipe can't /// stall it; log notable events for diagnostics. When `notices` is `Some`, each -/// parsed event is also forwarded to the caller (the core, which translates the -/// `app_audio` ones into a UI warning); a send failure (receiver dropped) just -/// stops forwarding, draining continues. The task ends on EOF (child exited). +/// parsed event is also forwarded to the caller (the core), and when the stream +/// ends — EOF or read error, i.e. the child exited or its event stream broke — +/// a final [`HostNotice::Eof`] is sent so the caller learns the child is gone +/// (a host that dies must not stay advertised as sharing). A send failure +/// (receiver dropped) just stops forwarding, draining continues. fn drain_in_background( mut lines: tokio::io::Lines>, role: &'static str, - notices: Option>, + notices: Option>, ) where R: tokio::io::AsyncRead + Unpin + Send + 'static, { @@ -587,10 +607,14 @@ fn drain_in_background( if let Some(ev) = parse_pixelpass_event(&line) { crate::log_msg(&format!("pixelpass {role}: {}", event_for_log(&ev))); if let Some(tx) = ¬ices { - let _ = tx.send(ev); + let _ = tx.send(HostNotice::Event(ev)); } } } + if let Some(tx) = ¬ices { + crate::log_msg(&format!("pixelpass {role}: stdout ended")); + let _ = tx.send(HostNotice::Eof); + } }); } @@ -1376,4 +1400,63 @@ Install hint: sudo apt install gstreamer1.0-plugins-bad #[cfg(not(windows))] assert_eq!(candidates, vec![dir.join("pixelpass")]); } + + /// The host-fault contract, clean-exit half: events are forwarded in order + /// and the stream ending yields exactly one terminal [`HostNotice::Eof`], + /// after which the drain task drops its sender (the closed channel is what + /// ends the core's forwarder). A host that dies silently — EOF swallowed — + /// is the S2 defect: the dead share stays advertised in presence. + #[tokio::test] + async fn drain_forwards_events_then_synthesizes_eof_when_stdout_ends() { + let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); + let (read_half, mut write_half) = tokio::io::duplex(1024); + drain_in_background(BufReader::new(read_half).lines(), "test", Some(tx)); + + use tokio::io::AsyncWriteExt; + write_half + .write_all(b"{\"event\":\"app_audio\",\"state\":\"routed\"}\nnot json\n") + .await + .unwrap(); + drop(write_half); // child exited: stdout EOF + + assert_eq!( + rx.recv().await, + Some(HostNotice::Event(PixelpassEvent::AppAudioRouted)) + ); + // The non-JSON line is dropped, not forwarded. + assert_eq!(rx.recv().await, Some(HostNotice::Eof)); + assert_eq!(rx.recv().await, None, "task ended and dropped the sender"); + } + + /// The host-fault contract, broken-stream half: a read *error* (not a tidy + /// EOF) must synthesize the same terminal `Eof` — the event stream is gone + /// either way, and only the drain task can tell the core so. + #[tokio::test] + async fn drain_synthesizes_eof_on_a_read_error_too() { + struct BrokenPipe; + impl tokio::io::AsyncRead for BrokenPipe { + fn poll_read( + self: std::pin::Pin<&mut Self>, + _cx: &mut std::task::Context<'_>, + _buf: &mut tokio::io::ReadBuf<'_>, + ) -> std::task::Poll> { + std::task::Poll::Ready(Err(std::io::Error::other("stream broke"))) + } + } + + use tokio::io::AsyncReadExt; + // One good event line, then the stream breaks mid-read. + let reader = + std::io::Cursor::new(b"{\"event\":\"capture\",\"state\":\"started\"}\n".to_vec()) + .chain(BrokenPipe); + let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); + drain_in_background(BufReader::new(reader).lines(), "test", Some(tx)); + + assert_eq!( + rx.recv().await, + Some(HostNotice::Event(PixelpassEvent::CaptureStarted)) + ); + assert_eq!(rx.recv().await, Some(HostNotice::Eof)); + assert_eq!(rx.recv().await, None, "task ended and dropped the sender"); + } }