From 7ea01a7e6768ae71090ddbfc6531ae1301bfa047 Mon Sep 17 00:00:00 2001 From: vicrodh <103399283+vicrodh@users.noreply.github.com> Date: Sun, 6 Sep 2026 05:05:27 -0600 Subject: [PATCH] fix(startup): bound and cancel the Linux instance probe Use one two-second deadline across native Unix connect, D-Bus authentication, name ownership and URL/window forwarding. Own and shut down the transport on cancellation, preserve unconfirmed links, and keep the primary server alive. Cover stalled and delayed phases, filesystem/abstract buses, ownership release and early UI handoff with isolated fixtures. All 733 Qt tests pass; async-io features remain unchanged. cx --- crates/Cargo.lock | 1 + crates/qbz-qt/Cargo.toml | 1 + crates/qbz-qt/src/deep_link_qt.rs | 8 + crates/qbz-qt/src/single_instance_qt.rs | 191 ++++++++-- crates/qbz-qt/src/single_instance_tests.rs | 420 +++++++++++++++++++++ 5 files changed, 585 insertions(+), 36 deletions(-) create mode 100644 crates/qbz-qt/src/single_instance_tests.rs diff --git a/crates/Cargo.lock b/crates/Cargo.lock index 516c3ad88..8368a9166 100644 --- a/crates/Cargo.lock +++ b/crates/Cargo.lock @@ -4522,6 +4522,7 @@ name = "qbz-qt" version = "2.1.0" dependencies = [ "arboard", + "async-io", "async-trait", "cc", "cxx", diff --git a/crates/qbz-qt/Cargo.toml b/crates/qbz-qt/Cargo.toml index 737750ad9..0678c4aa9 100644 --- a/crates/qbz-qt/Cargo.toml +++ b/crates/qbz-qt/Cargo.toml @@ -151,6 +151,7 @@ rfd = { version = "0.17", default-features = false, features = ["xdg-portal"] } # Feature unification is per-crate-graph, not per-target: any dependency that # turns on zbus/tokio changes the whole binary. [target.'cfg(target_os = "linux")'.dependencies] +async-io = "2" ksni = { version = "0.3", default-features = false, features = ["blocking", "async-io"] } # Session-bus single-instance guard. Keep the same async-io executor as ksni; # enabling zbus/tokio would poison the graph-wide executor choice. diff --git a/crates/qbz-qt/src/deep_link_qt.rs b/crates/qbz-qt/src/deep_link_qt.rs index 2af728993..517ac5ae8 100644 --- a/crates/qbz-qt/src/deep_link_qt.rs +++ b/crates/qbz-qt/src/deep_link_qt.rs @@ -113,6 +113,14 @@ pub(crate) fn stash(url: String) { } } +/// Return an unconfirmed handoff without replacing a newer launch intent. +#[cfg(target_os = "linux")] +pub(crate) fn restore_pending(url: String) { + if let Ok(mut pending) = PENDING.lock() { + pending.get_or_insert(url); + } +} + pub(crate) fn take_pending() -> Option { PENDING.lock().ok().and_then(|mut pending| pending.take()) } diff --git a/crates/qbz-qt/src/single_instance_qt.rs b/crates/qbz-qt/src/single_instance_qt.rs index de14a3916..30005f118 100644 --- a/crates/qbz-qt/src/single_instance_qt.rs +++ b/crates/qbz-qt/src/single_instance_qt.rs @@ -6,29 +6,34 @@ //! blocked by a broken or absent session bus. #![cfg(target_os = "linux")] +use std::cell::Cell; +use std::os::unix::net::UnixStream; use std::sync::atomic::{AtomicBool, Ordering}; -use std::sync::OnceLock; +use std::sync::{Arc, OnceLock}; +use std::time::{Duration, Instant}; -use zbus::blocking::fdo::DBusProxy; -use zbus::blocking::Connection; +use async_io::{Async, Timer}; +use futures_util::future::{select, Either}; use zbus::fdo::{RequestNameFlags, RequestNameReply}; -use zbus::names::WellKnownName; +use zbus::Connection; const BUS_NAME: &str = "com.blitzfc.qbz"; const OBJECT_PATH: &str = "/com/blitzfc/qbz"; const IFACE_NAME: &str = "com.blitzfc.qbz.SingleInstance"; -static CONN: OnceLock = OnceLock::new(); +const PROBE_TIMEOUT: Duration = Duration::from_secs(2); + +static CONN: OnceLock = OnceLock::new(); static UI_READY: AtomicBool = AtomicBool::new(false); static PENDING_PRESENT: AtomicBool = AtomicBool::new(false); struct SingleInstanceIface; fn present_or_defer() { - if UI_READY.load(Ordering::SeqCst) { + PENDING_PRESENT.store(true, Ordering::SeqCst); + // Publishing the intent first closes the race with bind_ui's drain. + if UI_READY.load(Ordering::SeqCst) && PENDING_PRESENT.swap(false, Ordering::SeqCst) { crate::tray_qt::present(); - } else { - PENDING_PRESENT.store(true, Ordering::SeqCst); } } @@ -54,42 +59,130 @@ pub(crate) fn bind_ui() { } } +/// Own the transport as well as zbus. Cancellation shuts down the socket +/// synchronously, even if an executor task still holds a connection clone. +/// Nothing can acquire a name or send another RPC after this guard is dropped. +struct SessionConnection { + socket: Arc>, + connection: Option, +} + +impl Drop for SessionConnection { + fn drop(&mut self) { + let _ = self.socket.get_ref().shutdown(std::net::Shutdown::Both); + } +} + /// True when this process should continue as the primary instance. +/// Uncertain IPC deliberately sacrifices uniqueness for availability. A link +/// is consumed only after a successful OpenUrl reply; we never retry a timed +/// out RPC. The remote may have acted before its reply was lost: exactly-once +/// delivery cannot be promised when the bus fails mid-call. pub(crate) fn acquire_or_raise() -> bool { - match probe() { - Ok(primary) => primary, - Err(e) => { - log::warn!("[qbz-qt] single-instance probe failed ({e}); continuing"); + let mut pending = crate::deep_link_qt::take_pending(); + let result = zbus::Address::session() + .map_err(|error| format!("address: {error}")) + .and_then(|address| probe_at(&address, &mut pending, PROBE_TIMEOUT, SingleInstanceIface)); + if let Some(url) = pending { + crate::deep_link_qt::restore_pending(url); + } + match result { + Ok(Some(conn)) => { + let _ = CONN.set(conn); + true + } + Ok(None) => false, + Err(error) => { + log::warn!("[qbz-qt] single-instance probe failed ({error}); continuing"); true } } } -fn probe() -> zbus::Result { - let conn = Connection::session()?; - conn.object_server().at(OBJECT_PATH, SingleInstanceIface)?; - let proxy = DBusProxy::new(&conn)?; - let name: WellKnownName<'_> = BUS_NAME.try_into().map_err(zbus::Error::from)?; - match proxy.request_name(name, RequestNameFlags::DoNotQueue.into())? { - RequestNameReply::PrimaryOwner | RequestNameReply::AlreadyOwner => { - let _ = CONN.set(conn); - Ok(true) +/// zbus's address connector uses a blocking worker for Unix connect. Use +/// async-io directly so a full listen backlog cannot leave a worker behind. +/// Native Linux session buses use filesystem or abstract Unix sockets. Other +/// transports fail open immediately: this optional startup guard must not +/// spawn an autolaunch/SSH command or uncancelable DNS lookup. +fn socket_path(address: &zbus::Address) -> zbus::Result { + use std::os::unix::ffi::{OsStrExt, OsStringExt}; + use zbus::address::transport::{Transport, UnixSocket}; + match address.transport() { + Transport::Unix(unix) => match unix.path() { + UnixSocket::File(path) => Ok(path.clone()), + UnixSocket::Abstract(name) => { + let mut bytes = vec![0]; + bytes.extend_from_slice(name.as_bytes()); + // async-io interprets a leading NUL as an abstract name. + Ok(std::ffi::OsString::from_vec(bytes).into()) + } + _ => Err(zbus::Error::Unsupported), + }, + _ => Err(zbus::Error::Unsupported), + } +} + +fn probe_at( + address: &zbus::Address, + pending: &mut Option, + budget: Duration, + interface: I, +) -> Result, String> { + let deadline = Instant::now() + budget; + let phase = Cell::new("connect"); + let operation = async { + let socket = Arc::new(Async::::connect(socket_path(address)?).await?); + let mut owned = SessionConnection { + socket, + connection: None, + }; + phase.set("AUTH/Hello/object registration"); + let conn = zbus::connection::Builder::socket(zbus::connection::socket::BoxedSplit::new( + Box::new(owned.socket.clone()), + Box::new(owned.socket.clone()), + )) + .serve_at(OBJECT_PATH, interface)? + .build() + .await?; + if address + .guid() + .is_some_and(|expected| *expected != **conn.server_guid()) + { + return Err(zbus::Error::Handshake("session bus GUID mismatch".into())); } - RequestNameReply::Exists | RequestNameReply::InQueue => { - let forwarded = match crate::deep_link_qt::take_pending() { - Some(url) => conn - .call_method( + owned.connection = Some(conn.clone()); + phase.set("RequestName"); + // Raw calls avoid proxy property subscriptions and detached cleanup + // RPCs. The sole timeout below includes every call and the handshake. + let reply = conn + .call_method( + Some("org.freedesktop.DBus"), + "/org/freedesktop/DBus", + Some("org.freedesktop.DBus"), + "RequestName", + &(BUS_NAME, RequestNameFlags::DoNotQueue as u32), + ) + .await?; + match reply.body().deserialize::()? { + RequestNameReply::PrimaryOwner | RequestNameReply::AlreadyOwner => Ok(Some(owned)), + RequestNameReply::Exists | RequestNameReply::InQueue => { + if let Some(url) = pending.as_deref() { + phase.set("OpenUrl"); + conn.call_method( Some(BUS_NAME), OBJECT_PATH, Some(IFACE_NAME), "OpenUrl", &url, ) - .is_ok(), - None => false, - }; - let presented = forwarded - || conn + .await?; + // No await after confirmation: no late completion can + // consume a link that the caller has already recovered. + pending.take(); + return Ok(None); + } + phase.set("Present"); + if conn .call_method( Some(BUS_NAME), OBJECT_PATH, @@ -97,17 +190,43 @@ fn probe() -> zbus::Result { "Present", &(), ) - .is_ok(); - if !presented { - let _ = conn.call_method( + .await + .is_ok() + { + return Ok(None); + } + phase.set("MPRIS Raise"); + conn.call_method( Some("org.mpris.MediaPlayer2.com.blitzfc.qbz"), "/org/mpris/MediaPlayer2", Some("org.mpris.MediaPlayer2"), "Raise", &(), - ); + ) + .await?; + Ok(None) } - Ok(false) } - } + }; + async_io::block_on(async { + // Timer is polled first, including when both futures become ready in + // the same turn. Dropping the losing future closes its socket guard. + match select(Box::pin(Timer::at(deadline)), Box::pin(operation)).await { + Either::Left((_, operation)) => { + drop(operation); + Err(format!( + "{}: total deadline of {} ms expired", + phase.get(), + budget.as_millis() + )) + } + Either::Right((result, _)) => { + result.map_err(|error: zbus::Error| format!("{}: {error}", phase.get())) + } + } + }) } + +#[cfg(test)] +#[path = "single_instance_tests.rs"] +mod tests; diff --git a/crates/qbz-qt/src/single_instance_tests.rs b/crates/qbz-qt/src/single_instance_tests.rs new file mode 100644 index 000000000..281b55468 --- /dev/null +++ b/crates/qbz-qt/src/single_instance_tests.rs @@ -0,0 +1,420 @@ +//! Synthetic buses use real wire messages, isolated sockets and no global env. +use super::*; +use std::io::{BufRead, BufReader, Read, Write}; +use std::os::unix::net::UnixListener; +use std::sync::Mutex; +use zbus::message::Message; +use zbus::zvariant::{serialized, Endian}; + +#[derive(Clone, Default)] +struct TestIface(Arc>>); + +#[zbus::interface(name = "com.blitzfc.qbz.SingleInstance")] +impl TestIface { + fn present(&self) { + self.0.lock().unwrap().push("present".into()); + } + fn open_url(&self, url: &str) { + self.0.lock().unwrap().push(url.into()); + } +} + +struct TempBusDir(std::path::PathBuf); +impl TempBusDir { + fn new() -> Self { + let path = std::env::temp_dir().join(format!("qbz bus {}", uuid::Uuid::new_v4())); + std::fs::create_dir(&path).unwrap(); + Self(path) + } + fn address(&self) -> zbus::Address { + use zbus::address::transport::{Transport, Unix, UnixSocket}; + zbus::Address::new(Transport::Unix(Unix::new(UnixSocket::File( + self.0.join("bus"), + )))) + } +} +impl Drop for TempBusDir { + fn drop(&mut self) { + let _ = std::fs::remove_dir_all(&self.0); + } +} + +fn read_message(reader: &mut BufReader) -> Message { + let mut bytes = vec![0; 16]; + reader.read_exact(&mut bytes).unwrap(); + let endian = if bytes[0] == b'l' { + Endian::Little + } else { + Endian::Big + }; + let number = |slice: &[u8]| { + let bytes = slice.try_into().unwrap(); + if endian == Endian::Little { + u32::from_le_bytes(bytes) + } else { + u32::from_be_bytes(bytes) + } + }; + let body = number(&bytes[4..8]) as usize; + let fields = number(&bytes[12..16]) as usize; + let size = (16 + fields + 7) / 8 * 8 + body; + assert!(size <= 65536); + bytes.resize(size, 0); + reader.read_exact(&mut bytes[16..]).unwrap(); + // SAFETY: bytes come exclusively from zbus's serializer in this test; + // no file descriptors are sent by any of these methods. + unsafe { + Message::from_bytes(serialized::Data::new( + bytes, + serialized::Context::new_dbus(endian, 0), + )) + } + .unwrap() +} + +fn reply( + reader: &mut BufReader, + message: &Message, + body: impl serde::Serialize + zbus::zvariant::DynamicType, +) { + let response = Message::method_return(&message.header()) + .unwrap() + .build(&body) + .unwrap(); + reader.get_mut().write_all(response.data()).unwrap(); +} + +fn reject(reader: &mut BufReader, message: &Message) { + let response = Message::error( + &message.header(), + "org.freedesktop.DBus.Error.UnknownMethod", + ) + .unwrap() + .build(&"fixture rejection") + .unwrap(); + reader.get_mut().write_all(response.data()).unwrap(); +} + +/// After the deadline, no bytes may follow and EOF must be observable. This +/// catches detached AUTH/RPC workers that a mere elapsed-time assertion misses. +fn assert_closed(reader: &mut BufReader) { + let mut byte = [0]; + assert_eq!( + reader.read(&mut byte).unwrap(), + 0, + "probe left a live socket or sent another call" + ); +} + +fn stalled_phase(phase: &'static str, with_url: bool) { + stalled_phase_after_delays(phase, with_url, Duration::ZERO, Duration::from_millis(150)); +} + +fn stalled_phase_after_delays( + phase: &'static str, + with_url: bool, + delay: Duration, + budget: Duration, +) { + let dir = TempBusDir::new(); + let listener = UnixListener::bind(dir.0.join("bus")).unwrap(); + listener.set_nonblocking(true).unwrap(); + let worker = std::thread::spawn(move || { + let start = Instant::now(); + let socket = loop { + match listener.accept() { + Ok((socket, _)) => break socket, + Err(e) + if e.kind() == std::io::ErrorKind::WouldBlock + && start.elapsed() < Duration::from_secs(2) => + { + std::thread::sleep(Duration::from_millis(2)) + } + Err(e) => panic!("fixture accept: {e}"), + } + }; + socket + .set_read_timeout(Some(Duration::from_secs(2))) + .unwrap(); + let mut reader = BufReader::new(socket); + let mut line = String::new(); + reader.read_line(&mut line).unwrap(); + assert!(line.contains("AUTH EXTERNAL")); + if phase == "AUTH" { + assert_closed(&mut reader); + return; + } + reader + .get_mut() + .write_all(b"OK 0123456789abcdef0123456789abcdef\r\n") + .unwrap(); + loop { + line.clear(); + reader.read_line(&mut line).unwrap(); + if line.starts_with("NEGOTIATE_UNIX_FD") { + reader.get_mut().write_all(b"AGREE_UNIX_FD\r\n").unwrap(); + } else if line.starts_with("BEGIN") { + break; + } else { + panic!("unexpected handshake: {line:?}"); + } + } + loop { + let message = read_message(&mut reader); + let header = message.header(); + let member = header.member().unwrap().as_str(); + if member == phase { + assert_closed(&mut reader); + // Even a reply arriving after cancellation cannot reactivate + // the caller or cause a follow-up RPC: its transport is shut. + let response = Message::method_return(&header).unwrap().build(&()).unwrap(); + let _ = reader.get_mut().write_all(response.data()); + return; + } + std::thread::sleep(delay); + match member { + "Hello" => reply(&mut reader, &message, ":1.42"), + "RequestName" => reply(&mut reader, &message, 3u32), + "Present" => reject(&mut reader, &message), + other => panic!("unexpected fixture method: {other}"), + } + } + }); + let mut url = with_url.then(|| "qobuzapp://album/fixture".to_owned()); + let before = url.clone(); + let started = Instant::now(); + let result = probe_at(&dir.address(), &mut url, budget, TestIface::default()); + let error = result.err().expect("stalled bus must fail open"); + assert!(error.contains("deadline"), "{error}"); + assert!( + started.elapsed() < budget + Duration::from_millis(170), + "{error}" + ); + if !delay.is_zero() { + assert!( + error.contains(phase), + "must reach the final phase within the global budget: {error}" + ); + } + assert_eq!(url, before, "unconfirmed link was lost"); + worker.join().unwrap(); +} + +#[test] +fn auth_timeout_closes_socket_repeatedly() { + for _ in 0..3 { + stalled_phase("AUTH", true); + } +} +#[test] +fn hello_timeout_closes_socket() { + stalled_phase("Hello", true); +} +#[test] +fn request_name_timeout_closes_socket() { + stalled_phase("RequestName", true); +} +#[test] +fn open_url_timeout_keeps_unconfirmed_link() { + stalled_phase("OpenUrl", true); +} +#[test] +fn present_timeout_closes_socket() { + stalled_phase("Present", false); +} +#[test] +fn raise_uses_the_same_deadline() { + stalled_phase("Raise", false); +} + +#[test] +fn slow_successful_calls_cannot_restart_the_final_rpc_budget() { + stalled_phase_after_delays( + "Raise", + false, + Duration::from_millis(80), + Duration::from_millis(350), + ); +} + +#[test] +fn missing_and_refused_bus_return_promptly() { + let dir = TempBusDir::new(); + for refused in [false, true] { + if refused { + drop(UnixListener::bind(dir.0.join("bus")).unwrap()); + } + let started = Instant::now(); + let mut link = Some("qobuzapp://track/1".into()); + assert!(probe_at( + &dir.address(), + &mut link, + PROBE_TIMEOUT, + TestIface::default() + ) + .is_err()); + assert!(started.elapsed() < Duration::from_secs(1)); + assert!(link.is_some()); + } +} + +#[test] +fn unsupported_transport_does_not_launch_an_external_worker() { + let address: zbus::Address = "tcp:host=unresolvable.invalid,port=1234" + .try_into() + .unwrap(); + assert!(socket_path(&address).is_err()); +} + +struct RealBus { + process: std::process::Child, + _dir: TempBusDir, + address: zbus::Address, +} +impl RealBus { + fn new() -> Self { + Self::with_abstract_socket(false) + } + fn with_abstract_socket(abstract_socket: bool) -> Self { + let dir = TempBusDir::new(); + let address = if abstract_socket { + format!("unix:abstract=qbz-test-{}", uuid::Uuid::new_v4()) + .as_str() + .try_into() + .unwrap() + } else { + dir.address() + }; + let mut process = std::process::Command::new("dbus-daemon") + .args(["--session", "--nofork", "--print-address=1"]) + .arg(format!("--address={address}")) + .stdout(std::process::Stdio::piped()) + .stderr(std::process::Stdio::null()) + .spawn() + .expect("dbus-daemon is required by the Linux Qt gate"); + let mut ready = String::new(); + BufReader::new(process.stdout.take().unwrap()) + .read_line(&mut ready) + .unwrap(); + if !ready.starts_with("unix:") { + let _ = process.kill(); + let _ = process.wait(); + panic!("private dbus-daemon did not publish its address"); + } + Self { + process, + _dir: dir, + address, + } + } +} +impl Drop for RealBus { + fn drop(&mut self) { + let _ = self.process.kill(); + let _ = self.process.wait(); + } +} + +#[test] +fn primary_keeps_server_alive_and_secondaries_forward_once() { + let bus = RealBus::new(); + let interface = TestIface::default(); + let primary = probe_at(&bus.address, &mut None, PROBE_TIMEOUT, interface.clone()) + .unwrap() + .unwrap(); + for link in [Some("qobuzapp://album/one".to_string()), None] { + let mut pending = link; + assert!(probe_at( + &bus.address, + &mut pending, + PROBE_TIMEOUT, + TestIface::default() + ) + .unwrap() + .is_none()); + assert!(pending.is_none()); + } + assert_eq!( + *interface.0.lock().unwrap(), + ["qobuzapp://album/one", "present"] + ); + drop(primary); + assert!( + probe_at(&bus.address, &mut None, PROBE_TIMEOUT, TestIface::default()) + .unwrap() + .is_some() + ); +} + +#[test] +fn native_abstract_session_bus_keeps_ownership() { + let bus = RealBus::with_abstract_socket(true); + let interface = TestIface::default(); + let primary = probe_at(&bus.address, &mut None, PROBE_TIMEOUT, interface.clone()) + .unwrap() + .unwrap(); + assert!( + probe_at(&bus.address, &mut None, PROBE_TIMEOUT, TestIface::default()) + .unwrap() + .is_none() + ); + assert_eq!(*interface.0.lock().unwrap(), ["present"]); + drop(primary); +} + +#[test] +fn early_handoff_survives_ui_binding_in_an_isolated_process() { + const CHILD: &str = "QBZ_TEST_EARLY_HANDOFF"; + if std::env::var(CHILD).as_deref() == Ok("1") { + // Production interface + pending-link/controller state, without any + // Qt window, authenticated session, keyring or global state of other + // tests. The actual Qt-thread hop is checked by the release smoke. + let interface = SingleInstanceIface; + interface.open_url("qobuzapp://album/before-bind"); + assert!(!UI_READY.load(Ordering::SeqCst)); + assert!(PENDING_PRESENT.load(Ordering::SeqCst)); + assert_eq!( + crate::deep_link_qt::take_pending().as_deref(), + Some("qobuzapp://album/before-bind") + ); + crate::deep_link_qt::restore_pending("qobuzapp://album/unconfirmed".into()); + crate::deep_link_qt::stash("qobuzapp://album/newer".into()); + crate::deep_link_qt::restore_pending("qobuzapp://album/older".into()); + bind_ui(); + assert!(UI_READY.load(Ordering::SeqCst)); + assert!(!PENDING_PRESENT.load(Ordering::SeqCst)); + assert_eq!( + crate::deep_link_qt::take_pending().as_deref(), + Some("qobuzapp://album/newer") + ); + for shown in [false, true] { + crate::tray_qt::set_window_shown(shown); + interface.present(); + assert!(!PENDING_PRESENT.load(Ordering::SeqCst)); + assert!(crate::deep_link_qt::take_pending().is_none()); + } + return; + } + let mut child = std::process::Command::new(std::env::current_exe().unwrap()) + .args([ + "--exact", + "single_instance_qt::tests::early_handoff_survives_ui_binding_in_an_isolated_process", + "--nocapture", + ]) + .env(CHILD, "1") + .spawn() + .unwrap(); + let started = Instant::now(); + loop { + if let Some(status) = child.try_wait().unwrap() { + assert!(status.success()); + break; + } + if started.elapsed() > Duration::from_secs(5) { + let _ = child.kill(); + let _ = child.wait(); + panic!("isolated handoff controller test did not terminate"); + } + std::thread::sleep(Duration::from_millis(10)); + } +}