diff --git a/libs/scrap/src/wayland/pipewire.rs b/libs/scrap/src/wayland/pipewire.rs index 34e89788d..7bdc74f7c 100644 --- a/libs/scrap/src/wayland/pipewire.rs +++ b/libs/scrap/src/wayland/pipewire.rs @@ -24,7 +24,7 @@ use lazy_static::lazy_static; use serde::{Deserialize, Serialize}; use base::platform::linux::CMD_SH; -use hbb_common::{bail, config, serde_json, tokio, ResultType}; +use hbb_common::{anyhow::anyhow, bail, config, serde_json, tokio, ResultType}; use super::capturable::PixelProvider; use super::capturable::{Capturable, Recorder}; @@ -264,11 +264,18 @@ pub struct PipeWireRecorder { saved_raw_data: Vec, // for faster compare and copy } +// Element creation fails the same way for a plugin that is not installed as for one that is +// broken, and the name is the only thing that tells a user which package to look at. +fn gst_element(name: &str) -> ResultType { + gst::ElementFactory::make(name, None) + .map_err(|_| anyhow!(stage_err("gst-plugin", "missing", name))) +} + impl PipeWireRecorder { pub fn new(capturable: PipeWireCapturable) -> ResultType { let pipeline = gst::Pipeline::new(None); - let src = gst::ElementFactory::make("pipewiresrc", None)?; + let src = gst_element("pipewiresrc")?; src.set_property("fd", &capturable.fd.as_raw_fd())?; src.set_property("path", &format!("{}", capturable.path))?; src.set_property("keepalive_time", &1_000.as_raw_fd())?; @@ -283,9 +290,9 @@ impl PipeWireRecorder { // "no more output formats" / not-negotiated (-4). videoconvert accepts any // system-memory video/x-raw format, widening negotiation so the portal can // settle on a format it can deliver via its SHM path. - let convert = gst::ElementFactory::make("videoconvert", None)?; + let convert = gst_element("videoconvert")?; - let sink = gst::ElementFactory::make("appsink", None)?; + let sink = gst_element("appsink")?; sink.set_property("drop", &true)?; sink.set_property("max-buffers", &1u32)?; @@ -464,11 +471,112 @@ impl Drop for PipeWireRecorder { } } +// The portal handshake is four sequential requests whose outcomes arrive as asynchronous +// `Response` signals, so where and why it failed is known only inside the signal handler. +// Recording it here, instead of collapsing every outcome into one `failed` flag, is what lets +// the app side name the real cause rather than guess it from the error text. +#[derive(Clone, Copy)] +enum PortalStage { + CreateSession = 1, + SelectDevices = 2, + SelectSources = 3, + Start = 4, +} + +impl PortalStage { + fn as_str(&self) -> &'static str { + match self { + Self::CreateSession => "create-session", + Self::SelectDevices => "select-devices", + Self::SelectSources => "select-sources", + Self::Start => "start", + } + } + + fn from_u8(v: u8) -> Self { + match v { + 2 => Self::SelectDevices, + 3 => Self::SelectSources, + 4 => Self::Start, + _ => Self::CreateSession, + } + } +} + +// `wl-stage:::`, parsed by `map_err_scrap` on the app side. The detail +// reaches the user through a `{}` placeholder in a translated string, so it must not bring +// braces, control characters or unbounded length of its own. +fn stage_err(stage: &str, kind: &str, detail: &str) -> String { + let detail: String = detail + .chars() + .map(|c| if c.is_control() { ' ' } else { c }) + .filter(|c| *c != '{' && *c != '}') + .take(200) + .collect(); + format!("wl-stage:{}:{}:{}", stage, kind, detail.trim()) +} + +fn dbus_detail(err: &dbus::Error) -> String { + match (err.name(), err.message()) { + (Some(name), Some(message)) if !name.is_empty() && !message.is_empty() => { + format!("{}: {}", name, message) + } + (Some(name), _) if !name.is_empty() => name.to_owned(), + (_, message) => message.unwrap_or_default().to_owned(), + } +} + +#[derive(Clone)] +struct PortalTrace { + failed: Arc, + reason: Arc>>, + // The stage whose `Response` we are still waiting for, so the polling loop can tell a + // non-interactive step apart from the one that waits for a human. + waiting_for: Arc, +} + +impl PortalTrace { + fn new() -> Self { + Self { + failed: Arc::new(AtomicBool::new(false)), + reason: Arc::new(Mutex::new(None)), + waiting_for: Arc::new(AtomicU8::new(PortalStage::CreateSession as u8)), + } + } + + fn fail(&self, stage: PortalStage, kind: &str, detail: &str) { + self.record(stage_err(stage.as_str(), kind, detail)); + self.failed.store(true, Ordering::SeqCst); + } + + // The first failure is the cause; whatever follows it is a consequence. + fn record(&self, tag: String) { + if let Ok(mut reason) = self.reason.lock() { + if reason.is_none() { + *reason = Some(tag); + } + } + } + + fn waiting(&self, stage: PortalStage) { + self.waiting_for.store(stage as u8, Ordering::SeqCst); + } + + fn waiting_stage(&self) -> PortalStage { + PortalStage::from_u8(self.waiting_for.load(Ordering::SeqCst)) + } + + fn take_reason(&self) -> Option { + self.reason.lock().ok().and_then(|mut r| r.take()) + } +} + fn handle_response( conn: &SyncConnection, path: dbus::Path<'static>, mut f: F, - failure_out: Arc, + trace: PortalTrace, + stage: PortalStage, ) -> Result where F: FnMut( @@ -491,18 +599,18 @@ where 0 => {} 1 => { warn!("DBus response: User cancelled interaction."); - failure_out.store(true, Ordering::SeqCst); + trace.fail(stage, "declined", ""); return true; } c => { warn!("DBus response: Unknown error, code: {}.", c); - failure_out.store(true, Ordering::SeqCst); + trace.fail(stage, "portal-error", &c.to_string()); return true; } } if let Err(err) = f(r, c, m) { warn!("Error requesting screen capture via dbus: {}", err); - failure_out.store(true, Ordering::SeqCst); + trace.fail(trace.waiting_stage(), "internal", &err.to_string()); } true }) @@ -638,15 +746,16 @@ pub fn request_remote_desktop( INIT = true; } } - let conn = SyncConnection::new_session()?; + let conn = SyncConnection::new_session() + .map_err(|e| anyhow!(stage_err("session-bus", "dbus", &dbus_detail(&e))))?; let portal = get_portal(&conn); let mut args: PropMap = HashMap::new(); let fd: Arc>> = Arc::new(Mutex::new(None)); let fd_res = fd.clone(); let streams: Arc>> = Arc::new(Mutex::new(Vec::new())); let streams_res = streams.clone(); - let failure = Arc::new(AtomicBool::new(false)); - let failure_res = failure.clone(); + let trace = PortalTrace::new(); + let trace_res = trace.clone(); let session: Arc>> = Arc::new(Mutex::new(None)); let session_res = session.clone(); let create_session_handle_token = "u1"; @@ -679,16 +788,19 @@ pub fn request_remote_desktop( fd.clone(), streams.clone(), session.clone(), - failure.clone(), + trace.clone(), is_support_restore_token, capture_cursor, ), - failure_res.clone(), + trace.clone(), + PortalStage::CreateSession, )?; if is_server_running() { - let _ = screencast_portal::create_session(&portal, args)?; + let _ = screencast_portal::create_session(&portal, args) + .map_err(|e| anyhow!(stage_err("create-session", "dbus", &dbus_detail(&e))))?; } else { - let _ = remote_desktop_portal::create_session(&portal, args)?; + let _ = remote_desktop_portal::create_session(&portal, args) + .map_err(|e| anyhow!(stage_err("create-session", "dbus", &dbus_detail(&e))))?; } // wait 3 minutes for user interaction @@ -699,13 +811,14 @@ pub fn request_remote_desktop( break; } - if failure_res.load(Ordering::SeqCst) { + if trace_res.failed.load(Ordering::SeqCst) { break; } } let fd_res = fd_res.lock().unwrap(); let streams_res = streams_res.lock().unwrap(); let session_res = session_res.lock().unwrap(); + let have_fd = fd_res.is_some(); if let Some(fd_res) = fd_res.clone() { if let Some(session) = session_res.clone() { @@ -720,14 +833,20 @@ pub fn request_remote_desktop( } } } - bail!("Failed to obtain screen capture. You may need to upgrade the PipeWire library for better compatibility. Please check https://github.com/rustdesk/rustdesk/issues/8600#issuecomment-2254720954 for more details.") + bail!(trace_res.take_reason().unwrap_or_else(|| { + if have_fd { + stage_err("streams", "empty", "") + } else { + stage_err(trace_res.waiting_stage().as_str(), "no-response", "") + } + })) } fn on_create_session_response( fd: Arc>>, streams: Arc>>, session: Arc>>>, - failure: Arc, + trace: PortalTrace, is_support_restore_token: bool, capture_cursor: bool, ) -> impl Fn( @@ -793,12 +912,14 @@ fn on_create_session_response( on_select_sources_response( fd.clone(), streams.clone(), - failure.clone(), + trace.clone(), ses.clone(), is_support_restore_token, ), - failure.clone(), + trace.clone(), + PortalStage::SelectSources, )?; + trace.waiting(PortalStage::SelectSources); let _ = portal.select_sources(ses.clone(), args)?; } else { // TODO: support persist_mode for remote_desktop_portal @@ -817,12 +938,14 @@ fn on_create_session_response( on_select_devices_response( fd.clone(), streams.clone(), - failure.clone(), + trace.clone(), ses.clone(), is_support_restore_token, ), - failure.clone(), + trace.clone(), + PortalStage::SelectDevices, )?; + trace.waiting(PortalStage::SelectDevices); let _ = portal.select_devices(ses.clone(), args)?; } @@ -833,7 +956,7 @@ fn on_create_session_response( fn on_select_devices_response( fd: Arc>>, streams: Arc>>, - failure: Arc, + trace: PortalTrace, session: dbus::Path<'static>, is_support_restore_token: bool, ) -> impl Fn( @@ -862,12 +985,14 @@ fn on_select_devices_response( on_select_sources_response( fd.clone(), streams.clone(), - failure.clone(), + trace.clone(), session.clone(), is_support_restore_token, ), - failure.clone(), + trace.clone(), + PortalStage::SelectSources, )?; + trace.waiting(PortalStage::SelectSources); let _ = portal.select_sources(session.clone(), args)?; Ok(()) @@ -877,7 +1002,7 @@ fn on_select_devices_response( fn on_select_sources_response( fd: Arc>>, streams: Arc>>, - failure: Arc, + trace: PortalTrace, session: dbus::Path<'static>, is_support_restore_token: bool, ) -> impl Fn( @@ -902,8 +1027,10 @@ fn on_select_sources_response( session.clone(), is_support_restore_token, ), - failure.clone(), + trace.clone(), + PortalStage::Start, )?; + trace.waiting(PortalStage::Start); if is_server_running() { let _ = screencast_portal::start(&portal, session.clone(), "", args)?; } else { @@ -1555,3 +1682,29 @@ fn sort_streams( *streams = sorted_streams; *shared_displays = sorted_shared_displays; } + +#[cfg(test)] +mod tests { + use super::stage_err; + + #[test] + fn stage_err_keeps_the_detail_safe_for_a_placeholder() { + assert_eq!( + stage_err("start", "declined", ""), + "wl-stage:start:declined:" + ); + // Braces of its own would break the placeholder lookup on the peer. + assert_eq!( + stage_err("create-session", "dbus", "org.freedesktop.{Error}"), + "wl-stage:create-session:dbus:org.freedesktop.Error" + ); + assert_eq!( + stage_err("select-sources", "internal", "one\ntwo"), + "wl-stage:select-sources:internal:one two" + ); + assert_eq!( + stage_err("start", "internal", &"x".repeat(300)), + format!("wl-stage:start:internal:{}", "x".repeat(200)) + ); + } +} diff --git a/src/server/wayland.rs b/src/server/wayland.rs index d48a215a4..e9b9d104c 100644 --- a/src/server/wayland.rs +++ b/src/server/wayland.rs @@ -21,6 +21,7 @@ lazy_static::lazy_static! { static ref CAP_DISPLAY_INFO: RwLock> = RwLock::new(HashMap::new()); static ref PIPEWIRE_INITIALIZED: RwLock = RwLock::new(false); static ref LOG_SCRAP_COUNT: Mutex = Mutex::new(0); + static ref LAST_STAGE_ERR: Mutex = Mutex::new(String::new()); static ref ACTIVE_DISPLAY_COUNT: RwLock = RwLock::new(0); } @@ -44,11 +45,23 @@ pub(super) fn decrement_active_display_count() -> usize { fn map_err_scrap(err: String) -> io::Error { // to-do: Handle error better, do not restart server + // Reached by the capture loop, which is what this crude self-heal was for. A no-reply + // during the portal handshake is tagged below and is reported instead of exiting: at + // login there is someone waiting to be told, and a portal that is slow to activate is + // not a reason to take the service down. if err.starts_with("Did not receive a reply") { log::error!("Fatal pipewire error, {}", &err); std::process::exit(-1); } + if let Some(tag) = err.strip_prefix(WAYLAND_STAGE_TAG) { + log_staged_once(&err); + return io::Error::new( + io::ErrorKind::Other, + staged_message(tag, is_ubuntu_before_21()), + ); + } + if DISTRO.name.to_uppercase() == "Ubuntu".to_uppercase() { if DISTRO.version_id < "21".to_owned() { io::Error::new(io::ErrorKind::Other, SCRAP_UBUNTU_HIGHER_REQUIRED) @@ -74,6 +87,28 @@ fn map_err_scrap(err: String) -> io::Error { } } +/// `Display::all` and `Capturer::new` reach the peer through `map_err_scrap`, but +/// `fill_displays` opens a portal session of its own and returns its error straight up, so a +/// tag has to be resolved here or it lands in the login dialog verbatim. +fn map_staged_err(err: anyhow::Error) -> anyhow::Error { + let text = err.to_string(); + match text.strip_prefix(WAYLAND_STAGE_TAG) { + Some(tag) => { + log_staged_once(&text); + anyhow::anyhow!(staged_message(tag, is_ubuntu_before_21())) + } + None => err, + } +} + +fn log_staged_once(err: &str) { + let mut last = LAST_STAGE_ERR.lock().unwrap(); + if *last != err { + log::error!("Wayland portal handshake failed: {}", err); + *last = err.to_owned(); + } +} + fn try_log(err: &String) { let mut lock_count = LOG_SCRAP_COUNT.lock().unwrap(); if *lock_count >= 1000000 { @@ -85,6 +120,138 @@ fn try_log(err: &String) { *lock_count += 1; } +// Translation keys, so the key itself is the English text: an older peer that has never heard +// of them falls back to displaying the key and still reads as a sentence. +const WAYLAND_DECLINED: &str = "The screen sharing request was declined on the remote device"; +const WAYLAND_NO_ANSWER: &str = + "No one responded to the screen sharing request on the remote device"; +const WAYLAND_PORTAL_ENDED: &str = "The XDG Desktop Portal ended the screen sharing request ({})"; +// The remedy the message it replaces used to carry, minus the link: this is the outcome +// rustdesk/rustdesk#8600 is about. +const WAYLAND_NO_SCREEN: &str = + "The XDG Desktop Portal returned no screen to capture, the PipeWire library may be too old"; +const WAYLAND_GST_MISSING: &str = "A GStreamer plugin needed for screen capture is missing ({})"; + +const WAYLAND_STAGE_TAG: &str = "wl-stage:"; + +// `translate()` on the peer strips the braces itself, so what goes on the wire is the key +// with the detail still *inside* the placeholder. +fn with_detail(key: &str, detail: &str) -> String { + key.replace("{}", &format!("{{{}}}", detail)) +} + +fn is_ubuntu_before_21() -> bool { + DISTRO.name.to_uppercase() == "Ubuntu".to_uppercase() && DISTRO.version_id < "21".to_owned() +} + +/// Maps a `::` tag from the portal handshake, see +/// `scrap::wayland::pipewire`, onto what to tell the peer. Everything the capture loop reports +/// carries no tag and keeps the legacy substring heuristics above. +fn staged_message(tag: &str, ubuntu_before_21: bool) -> String { + let mut parts = tag.splitn(3, ':'); + let stage = parts.next().unwrap_or_default(); + let kind = parts.next().unwrap_or_default(); + let detail = parts.next().unwrap_or_default().trim(); + + // An outcome that says something about the machine is what the Ubuntu branch was written + // for, so that branch keeps it. An outcome that says what a person did is a fact no distro + // check can improve on. + let of_the_machine = |msg: &str| { + if ubuntu_before_21 { + SCRAP_UBUNTU_HIGHER_REQUIRED.to_owned() + } else { + msg.to_owned() + } + }; + + match (stage, kind) { + (_, "declined") => WAYLAND_DECLINED.to_owned(), + (_, "portal-error") => with_detail(WAYLAND_PORTAL_ENDED, detail), + ("start", "no-response") => WAYLAND_NO_ANSWER.to_owned(), + ("streams", _) => of_the_machine(WAYLAND_NO_SCREEN), + ("gst-plugin", _) => of_the_machine(&with_detail(WAYLAND_GST_MISSING, detail)), + // Everything else is the portal not delivering, which is what this key already says -- + // and unlike a message of our own it carries the `systemctl --user restart` remedy. + _ => of_the_machine(SCRAP_XDP_PORTAL_UNAVAILABLE), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn staged_message_names_the_stage() { + let m = |tag| staged_message(tag, false); + assert_eq!(m("start:declined:"), WAYLAND_DECLINED); + assert_eq!(m("start:no-response:"), WAYLAND_NO_ANSWER); + assert_eq!(m("streams:empty:"), WAYLAND_NO_SCREEN); + } + + #[test] + fn a_portal_that_did_not_deliver_keeps_the_message_that_says_how_to_restart_it() { + let m = |tag| staged_message(tag, false); + assert_eq!( + m("create-session:dbus:org.freedesktop.DBus.Error.ServiceUnknown"), + SCRAP_XDP_PORTAL_UNAVAILABLE + ); + assert_eq!( + m("create-session:no-response:"), + SCRAP_XDP_PORTAL_UNAVAILABLE + ); + assert_eq!( + m("select-sources:internal:no session_handle"), + SCRAP_XDP_PORTAL_UNAVAILABLE + ); + // A tag this build does not know must never fall back to a guess. + assert_eq!( + m("some-new-stage:some-new-kind:x"), + SCRAP_XDP_PORTAL_UNAVAILABLE + ); + assert_eq!(m(""), SCRAP_XDP_PORTAL_UNAVAILABLE); + } + + // The peer resolves a message by replacing its first `{...}` with `{}` and looking that + // up, so every detail-carrying message has to reduce back to its key exactly. + #[test] + fn a_detail_carrying_message_reduces_back_to_its_key() { + let reduce = |s: &str| { + let open = s.find('{').expect("no placeholder"); + let close = s[open..].find('}').expect("unclosed placeholder") + open; + format!("{}{{}}{}", &s[..open], &s[close + 1..]) + }; + let ended = staged_message("start:portal-error:2", false); + assert_eq!( + ended, + "The XDG Desktop Portal ended the screen sharing request ({2})" + ); + assert_eq!(reduce(&ended), WAYLAND_PORTAL_ENDED); + + let gst = staged_message("gst-plugin:missing:pipewiresrc", false); + assert_eq!( + gst, + "A GStreamer plugin needed for screen capture is missing ({pipewiresrc})" + ); + assert_eq!(reduce(&gst), WAYLAND_GST_MISSING); + } + + #[test] + fn legacy_ubuntu_keeps_its_message_for_machine_faults_only() { + let m = |tag| staged_message(tag, true); + assert_eq!( + m("create-session:dbus:org.freedesktop.DBus.Error.ServiceUnknown"), + SCRAP_UBUNTU_HIGHER_REQUIRED + ); + assert_eq!( + m("gst-plugin:missing:pipewiresrc"), + SCRAP_UBUNTU_HIGHER_REQUIRED + ); + assert_eq!(m("streams:empty:"), SCRAP_UBUNTU_HIGHER_REQUIRED); + assert_eq!(m("start:declined:"), WAYLAND_DECLINED); + assert_eq!(m("start:no-response:"), WAYLAND_NO_ANSWER); + } +} + struct CapturerPtr(*mut Capturer); impl Clone for CapturerPtr { @@ -312,7 +479,8 @@ pub(super) async fn check_init() -> ResultType<()> { { let temp_mouse_move_handle = input_service::TemporaryMouseMoveHandle::new(); let move_mouse_to = |x, y| temp_mouse_move_handle.move_mouse_to(x, y); - fill_displays(move_mouse_to, crate::get_cursor_pos, &mut all)?; + fill_displays(move_mouse_to, crate::get_cursor_pos, &mut all) + .map_err(map_staged_err)?; } log::debug!("Attempting to fix logical size with try_fix_logical_size()"); try_fix_logical_size(&mut all); @@ -340,10 +508,9 @@ pub(super) async fn check_init() -> ResultType<()> { // Create individual CapDisplayInfo for each display with its own capturer for (idx, display) in all.into_iter().enumerate() { - let capturer = - Box::into_raw(Box::new(Capturer::new(display).with_context(|| { - format!("Failed to create capturer for display {}", idx) - })?)); + // No `with_context` here: the peer is shown `format!("{}", err)`, which + // renders only the outermost layer, and the mapped reason is the inner one. + let capturer = Box::into_raw(Box::new(Capturer::new(display)?)); let capturer = CapturerPtr(capturer); let cap_display_info = Box::into_raw(Box::new(CapDisplayInfo {