general improvement and bug fixes, cursor scaling, DMA-BUF routing and nested backend recovery

ignore performance-baseline.md

more improvement and code quality fixes

portal/src/backend.rs

@@ -8,7 +8,7 @@ use std::time::Duration;
 
 use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
 use tokio::process::{Child, Command};
-use tokio::sync::{Mutex, Notify};
+use tokio::sync::{Mutex, Notify, OwnedSemaphorePermit, Semaphore};
 use zbus::Connection;
 use zbus::message::Header;
 use zbus::object_server::SignalEmitter;
@@ -22,6 +22,7 @@ pub(crate) type Options = HashMap<String, OwnedValue>;
 pub(crate) static CONFIG_TRANSACTION: std::sync::Mutex<()> = std::sync::Mutex::new(());
 const FRONTEND: &str = "org.freedesktop.portal.Desktop";
 const PATH: &str = "/org/freedesktop/portal/desktop";
+const MAX_SESSIONS: usize = 8;
 
 #[derive(Debug)]
 enum StartError {
@@ -82,11 +83,22 @@ struct Session {
     persist_mode: u32,
     restore: Option<crate::restore::Restore>,
     cancel: Arc<Cancel>,
+    work: Arc<OwnedSemaphorePermit>,
 }
 
-#[derive(Clone, Default)]
+#[derive(Clone)]
 pub struct Backend {
     sessions: Arc<Mutex<HashMap<String, Session>>>,
+    work: Arc<Semaphore>,
+}
+
+impl Default for Backend {
+    fn default() -> Self {
+        Self {
+            sessions: Arc::default(),
+            work: Arc::new(Semaphore::new(MAX_SESSIONS)),
+        }
+    }
 }
 
 // This private interface only transfers stop controls to the shell's own recorder.
@@ -252,11 +264,17 @@ impl Backend {
             return Err(error("Invalid session"));
         }
         let mut sessions = self.sessions.lock().await;
-        if sessions.len() >= 8 || sessions.contains_key(session_handle.as_str()) {
+        if sessions.len() >= MAX_SESSIONS || sessions.contains_key(session_handle.as_str()) {
             return Err(zbus::fdo::Error::LimitsExceeded(
                 "Session limit reached or duplicate session".into(),
             ));
         }
+        let work = Arc::new(
+            self.work
+                .clone()
+                .try_acquire_owned()
+                .map_err(|_| zbus::fdo::Error::LimitsExceeded("Capture work is still shutting down".into()))?,
+        );
         let inserted = connection
             .object_server()
             .at(
@@ -284,6 +302,7 @@ impl Backend {
                 persist_mode: 0,
                 restore: None,
                 cancel: Arc::default(),
+                work,
             },
         );
         Ok((0, Options::new()))
@@ -347,7 +366,7 @@ impl Backend {
             return Err(error("Invalid request path"));
         }
 
-        let (multiple, cursor, source_types, bar_controlled, persist_mode, restore, cancel) = {
+        let (multiple, cursor, source_types, bar_controlled, persist_mode, restore, cancel, work) = {
             let mut sessions = self.sessions.lock().await;
             let session = sessions
                 .get_mut(session_handle.as_str())
@@ -366,6 +385,7 @@ impl Backend {
                 session.persist_mode,
                 session.restore.clone(),
                 session.cancel.clone(),
+                session.work.clone(),
             )
         };
         let inserted = connection
@@ -400,6 +420,8 @@ impl Backend {
                 requested_persistence: persist_mode,
                 restore,
                 bar_controlled,
+                cancel: cancel.clone(),
+                work,
             },
         );
         let result = tokio::select! {
@@ -543,6 +565,39 @@ struct StreamOptions {
     requested_persistence: u32,
     restore: Option<crate::restore::Restore>,
     bar_controlled: bool,
+    cancel: Arc<Cancel>,
+    work: Arc<OwnedSemaphorePermit>,
+}
+
+async fn capture_work<T: Send + 'static>(
+    cancel: Arc<Cancel>,
+    permit: Arc<OwnedSemaphorePermit>,
+    operation: impl FnOnce(&AtomicBool) -> Result<T, String> + Send + 'static,
+) -> Result<T, String> {
+    tokio::task::spawn_blocking(move || {
+        // A dropped async future does not stop spawn_blocking. This lease
+        // stays with the worker until its connection and work are released.
+        let _permit = permit;
+        if cancel.stopped.load(Ordering::SeqCst) {
+            return Err("Capture cancelled".into());
+        }
+
+        operation(&cancel.stopped)
+    })
+    .await
+    .map_err(|error| error.to_string())?
+}
+
+async fn capture_ipc(
+    cancel: &Arc<Cancel>,
+    work: &Arc<OwnedSemaphorePermit>,
+    command: &'static str,
+    args: serde_json::Value,
+) -> Result<serde_json::Value, String> {
+    capture_work(cancel.clone(), work.clone(), move |_| {
+        crate::desktop::ipc_sync(command, args)
+    })
+    .await
 }
 
 async fn start_streams(
@@ -557,20 +612,20 @@ async fn start_streams(
         requested_persistence,
         restore,
         bar_controlled,
+        cancel,
+        work,
     } = options;
-    let mut discovery = tokio::task::spawn_blocking(|| Capture::connect(&AtomicBool::new(false)))
-        .await
-        .map_err(|e| e.to_string())??;
+    let mut discovery = capture_work(cancel.clone(), work.clone(), Capture::connect).await?;
     let mut sources = if source_types & 1 != 0 {
         discovery.sources()
     } else {
         Vec::new()
     };
     if source_types & 2 != 0 && discovery.supports_windows() {
-        let windows = crate::desktop::ipc("get-windows", serde_json::json!({})).await?;
+        let windows = capture_ipc(&cancel, &work, "get-windows", serde_json::json!({})).await?;
         sources.extend(crate::desktop::sharing_window_sources(&windows)?);
     }
-    let outputs = crate::desktop::ipc("get-outputs", serde_json::json!({})).await?;
+    let outputs = capture_ipc(&cancel, &work, "get-outputs", serde_json::json!({})).await?;
     let rememberable = sources
         .iter()
         .filter(|source| {
@@ -616,13 +671,21 @@ async fn start_streams(
             return Err("Invalid persistence selection".into());
         }
         let selected = validate_selection(&prompt.sources, &selection.names, multiple)?;
-        crate::desktop::wait_for_surface_removal(picker_pid).await?;
+        crate::desktop::wait_for_surface_removal_using(|| {
+            capture_ipc(
+                &cancel,
+                &work,
+                "has-client-surfaces",
+                serde_json::json!({"pid": picker_pid}),
+            )
+        })
+        .await?;
 
         (selected, selection.persist_mode)
     };
 
     let mut reply = Options::new();
-    let fresh_outputs = crate::desktop::ipc("get-outputs", serde_json::json!({})).await?;
+    let fresh_outputs = capture_ipc(&cancel, &work, "get-outputs", serde_json::json!({})).await?;
 
     let monitors = selected
         .iter()
@@ -698,11 +761,12 @@ async fn start_streams(
         children.push(child);
     }
 
-    tokio::task::spawn_blocking(move || discovery.validate_sources(&generations))
-        .await
-        .map_err(|error| error.to_string())??;
+    capture_work(cancel.clone(), work.clone(), move |stop| {
+        discovery.validate_sources(&generations, stop)
+    })
+    .await?;
 
-    let final_outputs = crate::desktop::ipc("get-outputs", serde_json::json!({})).await?;
+    let final_outputs = capture_ipc(&cancel, &work, "get-outputs", serde_json::json!({})).await?;
     if !crate::restore::selection_unchanged(&monitors, &outputs, &final_outputs) {
         return Err("A selected display changed during stream startup".into());
     }
@@ -711,7 +775,7 @@ async fn start_streams(
         return Err("A selected stream ended during startup".into());
     }
 
-    let final_windows = crate::desktop::ipc("get-windows", serde_json::json!({})).await?;
+    let final_windows = capture_ipc(&cancel, &work, "get-windows", serde_json::json!({})).await?;
     let streams = streams
         .into_iter()
         .map(|(name, ready)| {
@@ -914,6 +978,61 @@ pub async fn run() -> Result<(), Box<dyn std::error::Error>> {
 #[cfg(test)]
 mod tests {
     use super::*;
+
+    #[tokio::test]
+    async fn cancelled_async_work_holds_session_capacity_until_worker_exits() {
+        let backend = Backend::default();
+        let cancel = Arc::new(Cancel::default());
+        let permits: Vec<_> = (0..MAX_SESSIONS)
+            .map(|_| Arc::new(backend.work.clone().try_acquire_owned().unwrap()))
+            .collect();
+        let (started, received) = tokio::sync::oneshot::channel();
+        let (release, blocked) = std::sync::mpsc::channel();
+        let permit = permits[0].clone();
+        let flag = cancel.clone();
+        let task = tokio::spawn(async move {
+            capture_work(flag, permit, move |stop| {
+                started.send(()).unwrap();
+                blocked.recv_timeout(Duration::from_secs(2)).unwrap();
+                assert!(stop.load(Ordering::SeqCst));
+                Ok(())
+            })
+            .await
+        });
+        received.await.unwrap();
+        cancel.stop();
+        task.abort();
+        assert!(task.await.unwrap_err().is_cancelled());
+        // End the corresponding session while its blocking worker is alive.
+        let mut permits = permits;
+        permits.remove(0);
+        assert!(backend.work.clone().try_acquire_owned().is_err());
+        release.send(()).unwrap();
+        let recovered = tokio::time::timeout(Duration::from_secs(1), backend.work.clone().acquire_owned())
+            .await
+            .unwrap()
+            .unwrap();
+        assert_eq!(recovered.num_permits(), 1);
+        assert_eq!(backend.work.available_permits(), 0);
+        drop(recovered);
+        drop(permits);
+        assert_eq!(backend.work.available_permits(), MAX_SESSIONS);
+    }
+
+    #[tokio::test]
+    async fn already_cancelled_capture_never_starts_blocking_discovery() {
+        let backend = Backend::default();
+        let cancel = Arc::new(Cancel::default());
+        cancel.stop();
+        let permit = Arc::new(backend.work.clone().try_acquire_owned().unwrap());
+        let result = capture_work(cancel, permit, |_| -> Result<(), String> {
+            panic!("Cancelled discovery must not start");
+        })
+        .await;
+        assert!(result.unwrap_err().contains("cancelled"));
+        assert_eq!(backend.work.available_permits(), MAX_SESSIONS);
+    }
+
     #[test]
     fn bar_control_claims_are_bound_to_the_recorder_connection() {
         assert!(owns_session(

portal/src/bridge.rs

@@ -1,15 +1,25 @@
 use std::os::unix::net::UnixStream;
 use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, Ordering};
 use std::time::Duration;
 
 use serde_json::Value as Json;
+use tokio::sync::Notify;
 
 #[derive(Clone)]
 pub(crate) struct Bridge(Arc<BridgeInner>);
 
+struct Connection {
+    stream: UnixStream,
+    id: u64,
+}
+
 struct BridgeInner {
-    stream: std::sync::Mutex<UnixStream>,
+    connection: std::sync::Mutex<Connection>,
     shutdown: UnixStream,
+    closed: AtomicBool,
+    changed: Notify,
+    timeout: Duration,
 }
 
 impl Bridge {
@@ -19,19 +29,32 @@ impl Bridge {
             .ok_or("Missing runtime directory")?
             .join("ferese/control.sock");
         let stream = UnixStream::connect(path).map_err(|error| error.to_string())?;
+        Self::from_stream(stream, Duration::from_secs(2))
+    }
+
+    fn from_stream(stream: UnixStream, timeout: Duration) -> Result<Self, String> {
         stream
-            .set_read_timeout(Some(Duration::from_secs(2)))
+            .set_read_timeout(Some(timeout))
             .map_err(|error| error.to_string())?;
         stream
-            .set_write_timeout(Some(Duration::from_secs(2)))
+            .set_write_timeout(Some(timeout))
             .map_err(|error| error.to_string())?;
         let shutdown = stream.try_clone().map_err(|error| error.to_string())?;
         Ok(Self(Arc::new(BridgeInner {
-            stream: std::sync::Mutex::new(stream),
+            connection: std::sync::Mutex::new(Connection { stream, id: 0 }),
             shutdown,
+            closed: AtomicBool::new(false),
+            changed: Notify::new(),
+            timeout,
         })))
     }
 
+    #[cfg(test)]
+    pub(crate) fn test_pair() -> (Self, UnixStream) {
+        let (stream, peer) = UnixStream::pair().unwrap();
+        (Self::from_stream(stream, Duration::from_secs(1)).unwrap(), peer)
+    }
+
     pub(crate) async fn call(&self, command: &'static str, args: Json) -> Result<Json, String> {
         self.request(command, args, false).await
     }
@@ -43,28 +66,68 @@ impl Bridge {
 
     async fn request(&self, command: &'static str, args: Json, wait: bool) -> Result<Json, String> {
         let bridge = self.clone();
-        tokio::task::spawn_blocking(move || {
-            let mut stream = bridge.0.stream.lock().map_err(|_| "Compositor connection failed")?;
-            stream
-                .set_read_timeout(if wait { None } else { Some(Duration::from_secs(2)) })
-                .map_err(|error| error.to_string())?;
+        match tokio::task::spawn_blocking(move || bridge.request_sync(command, args, wait)).await {
+            Ok(result) => result,
+            Err(error) => {
+                self.close();
+                Err(error.to_string())
+            }
+        }
+    }
+
+    fn request_sync(&self, command: &'static str, args: Json, wait: bool) -> Result<Json, String> {
+        let mut connection = match self.0.connection.lock() {
+            Ok(connection) => connection,
+            Err(_) => {
+                self.close();
+                return Err("Compositor connection failed".into());
+            }
+        };
+
+        let response = (|| {
+            if self.is_closed() {
+                return Err("Compositor connection is closed".to_owned());
+            }
+
+            connection.id = connection.id.checked_add(1).ok_or("Compositor request IDs exhausted")?;
             let request = ferese_ipc::Request {
                 version: ferese_ipc::VERSION,
-                id: 1,
+                id: connection.id,
                 kind: "command".into(),
                 command: command.into(),
                 args,
             };
-            ferese_ipc::write_frame(&mut *stream, &request).map_err(|error| error.to_string())?;
-            let response: ferese_ipc::Response =
-                ferese_ipc::read_frame(&mut *stream).map_err(|error| error.to_string())?;
-            if let Some(error) = response.error {
-                return Err(error.message);
+            connection
+                .stream
+                .set_read_timeout(if wait { None } else { Some(self.0.timeout) })
+                .map_err(|error| error.to_string())?;
+            ferese_ipc::write_frame(&mut connection.stream, &request).map_err(|error| error.to_string())?;
+            let value: Json = ferese_ipc::read_frame(&mut connection.stream).map_err(|error| error.to_string())?;
+            let has_result = value.get("result").is_some();
+            let response: ferese_ipc::Response = serde_json::from_value(value).map_err(|error| error.to_string())?;
+            if response.id != request.id || response.version != ferese_ipc::VERSION {
+                return Err("Unexpected compositor response ID or protocol version".into());
             }
-            response.result.ok_or("Missing compositor response".into())
-        })
-        .await
-        .map_err(|error| error.to_string())?
+
+            if has_result == response.error.is_some() {
+                return Err("Invalid compositor response outcome".into());
+            }
+
+            Ok(response)
+        })();
+
+        match response {
+            Ok(response) => match response.error {
+                Some(error) => Err(error.message),
+                None => Ok(response.result.unwrap_or(Json::Null)),
+            },
+            Err(error) => {
+                // Poison the stream before releasing the request lock so a
+                // queued caller cannot read a late or partial response.
+                self.close();
+                Err(error)
+            }
+        }
     }
 
     pub(crate) async fn capture_watch(&self, session: u64) -> Result<Json, String> {
@@ -72,24 +135,52 @@ impl Bridge {
             .await
     }
 
+    pub(crate) fn is_closed(&self) -> bool {
+        self.0.closed.load(Ordering::SeqCst)
+    }
+
+    pub(crate) async fn terminated(&self) {
+        let changed = self.0.changed.notified();
+        tokio::pin!(changed);
+        changed.as_mut().enable();
+        if !self.is_closed() {
+            changed.await;
+        }
+    }
+
     pub(crate) async fn closed(&self) {
-        use std::io::Read;
         let Ok(reader) = self.0.shutdown.try_clone() else {
             return;
         };
-        if reader.set_nonblocking(true).is_err() {
-            return;
-        }
         let Ok(reader) = tokio::io::unix::AsyncFd::new(reader) else {
             return;
         };
         loop {
-            let Ok(mut ready) = reader.readable().await else {
+            let readable = tokio::select! {
+                _ = self.terminated() => return,
+                ready = reader.readable() => ready,
+            };
+            let Ok(mut ready) = readable else {
                 return;
             };
             match ready.try_io(|reader| {
-                let mut socket = reader.get_ref();
-                socket.read(&mut [0_u8; 1])
+                use std::os::fd::AsRawFd;
+                let mut byte = 0u8;
+                // Per-call nonblocking peek leaves the shared descriptor flags
+                // and response bytes untouched for the request reader.
+                let result = unsafe {
+                    libc::recv(
+                        reader.get_ref().as_raw_fd(),
+                        (&mut byte as *mut u8).cast(),
+                        1,
+                        libc::MSG_PEEK | libc::MSG_DONTWAIT,
+                    )
+                };
+                if result < 0 {
+                    Err(std::io::Error::last_os_error())
+                } else {
+                    Ok(result)
+                }
             }) {
                 Ok(_) => return,
                 Err(_) => continue,
@@ -98,6 +189,11 @@ impl Bridge {
     }
 
     pub(crate) fn close(&self) {
+        self.0.closed.store(true, Ordering::SeqCst);
         let _ = self.0.shutdown.shutdown(std::net::Shutdown::Both);
+        self.0.changed.notify_waiters();
     }
 }
+
+#[cfg(test)]
+mod tests;

portal/src/bridge/tests.rs

@@ -0,0 +1,158 @@
+use super::*;
+use ferese_ipc::{Request, Response, read_frame, write_frame};
+use serde_json::json;
+use std::io::{Read, Write};
+
+fn pair() -> (Bridge, UnixStream) {
+    pair_with_timeout(Duration::from_secs(1))
+}
+
+fn pair_with_timeout(timeout: Duration) -> (Bridge, UnixStream) {
+    let (client, server) = UnixStream::pair().unwrap();
+    server.set_read_timeout(Some(Duration::from_secs(2))).unwrap();
+    (Bridge::from_stream(client, timeout).unwrap(), server)
+}
+
+#[tokio::test]
+async fn valid_errors_preserve_connection_and_request_ids_are_unique() {
+    let (bridge, mut server) = pair();
+    let peer = std::thread::spawn(move || {
+        let first: Request = read_frame(&mut server).unwrap();
+        write_frame(&mut server, &Response::error(first.id, "denied", "Denied")).unwrap();
+        let second: Request = read_frame(&mut server).unwrap();
+        assert_eq!(second.id, first.id + 1);
+        write_frame(&mut server, &Response::success(second.id, json!({"enabled":false}))).unwrap();
+        let third: Request = read_frame(&mut server).unwrap();
+        assert_eq!(third.id, second.id + 1);
+        write_frame(&mut server, &Response::success(third.id, Json::Null)).unwrap();
+    });
+    assert_eq!(bridge.call("enable", json!({})).await.unwrap_err(), "Denied");
+    assert!(!bridge.is_closed());
+    assert_eq!(
+        bridge.call("disable", json!({})).await.unwrap(),
+        json!({"enabled":false})
+    );
+    assert!(!bridge.is_closed());
+    assert_eq!(bridge.call("empty-result", json!({})).await.unwrap(), Json::Null);
+    peer.join().unwrap();
+}
+
+#[tokio::test]
+async fn timeout_closes_stream_before_a_late_reply_can_be_reused() {
+    let (bridge, mut server) = pair_with_timeout(Duration::from_millis(40));
+    let (release, blocked) = std::sync::mpsc::channel();
+    let peer = std::thread::spawn(move || {
+        let request: Request = read_frame(&mut server).unwrap();
+        blocked.recv_timeout(Duration::from_secs(2)).unwrap();
+        assert!(write_frame(&mut server, &Response::success(request.id, json!({"enabled":true}))).is_err());
+        assert_eq!(server.read(&mut [0u8; 1]).unwrap(), 0);
+    });
+    assert!(bridge.call("enable", json!({})).await.is_err());
+    assert!(bridge.is_closed());
+    tokio::time::timeout(Duration::from_secs(1), bridge.terminated())
+        .await
+        .unwrap();
+    assert!(bridge.call("disable", json!({})).await.unwrap_err().contains("closed"));
+    release.send(()).unwrap();
+    peer.join().unwrap();
+}
+
+#[tokio::test]
+async fn partial_reply_timeout_cannot_corrupt_the_next_frame() {
+    let (bridge, mut server) = pair_with_timeout(Duration::from_millis(40));
+    let (release, blocked) = std::sync::mpsc::channel();
+    let peer = std::thread::spawn(move || {
+        let _: Request = read_frame(&mut server).unwrap();
+        server.write_all(&20u32.to_be_bytes()).unwrap();
+        server.write_all(b"{\"id\":").unwrap();
+        blocked.recv_timeout(Duration::from_secs(2)).unwrap();
+        assert_eq!(server.read(&mut [0u8; 1]).unwrap(), 0);
+    });
+    assert!(bridge.call("enable", json!({})).await.is_err());
+    assert!(bridge.is_closed());
+    assert!(bridge.call("disable", json!({})).await.is_err());
+    release.send(()).unwrap();
+    peer.join().unwrap();
+}
+
+#[tokio::test]
+async fn malformed_or_mismatched_replies_poison_the_bridge() {
+    for case in 0..6 {
+        let (bridge, mut server) = pair();
+        let peer = std::thread::spawn(move || {
+            let request: Request = read_frame(&mut server).unwrap();
+            let mut response = Response::success(request.id, json!({}));
+            match case {
+                0 => response.id += 1,
+                1 => response.version += 1,
+                2 => response.result = None,
+                3 => response.error = Response::error(request.id, "failed", "Failed").error,
+                4 => {
+                    server.write_all(&1u32.to_be_bytes()).unwrap();
+                    server.write_all(b"{").unwrap();
+                    return;
+                }
+                _ => {
+                    server.write_all(&u32::MAX.to_be_bytes()).unwrap();
+                    return;
+                }
+            }
+            write_frame(&mut server, &response).unwrap();
+        });
+        assert!(bridge.call("enable", json!({})).await.is_err(), "case {case}");
+        assert!(bridge.is_closed(), "case {case}");
+        assert!(bridge.call("disable", json!({})).await.is_err());
+        peer.join().unwrap();
+    }
+}
+
+#[tokio::test]
+async fn closing_any_clone_interrupts_a_blocked_watch() {
+    let (bridge, mut server) = pair();
+    let (started, received) = tokio::sync::oneshot::channel();
+    let peer = std::thread::spawn(move || {
+        let request: Request = read_frame(&mut server).unwrap();
+        assert_eq!(request.command, "input-capture-watch");
+        started.send(()).unwrap();
+        assert_eq!(server.read(&mut [0u8; 1]).unwrap(), 0);
+    });
+    let watcher = bridge.clone();
+    let task = tokio::spawn(async move { watcher.capture_watch(7).await });
+    received.await.unwrap();
+    bridge.close();
+    assert!(
+        tokio::time::timeout(Duration::from_secs(1), task)
+            .await
+            .unwrap()
+            .unwrap()
+            .is_err()
+    );
+    assert!(bridge.call("enable", json!({})).await.is_err());
+    peer.join().unwrap();
+}
+
+#[tokio::test]
+async fn peer_watch_keeps_descriptor_flags_and_response_bytes_untouched() {
+    use std::os::fd::AsRawFd;
+
+    let (bridge, mut server) = pair();
+    let flags = unsafe { libc::fcntl(bridge.0.shutdown.as_raw_fd(), libc::F_GETFL) };
+    server.write_all(b"x").unwrap();
+    tokio::time::timeout(Duration::from_secs(1), bridge.closed())
+        .await
+        .unwrap();
+    assert_eq!(
+        unsafe { libc::fcntl(bridge.0.shutdown.as_raw_fd(), libc::F_GETFL) },
+        flags
+    );
+    let mut byte = [0u8; 1];
+    bridge
+        .0
+        .connection
+        .lock()
+        .unwrap()
+        .stream
+        .read_exact(&mut byte)
+        .unwrap();
+    assert_eq!(byte, *b"x");
+}

portal/src/capture.rs

@@ -87,6 +87,10 @@ pub struct Capture {
 impl Capture {
     pub fn connect(stop: &AtomicBool) -> Result<Self, String> {
         let connection = Connection::connect_to_env().map_err(|e| e.to_string())?;
+        Self::connect_with_connection(connection, stop)
+    }
+
+    fn connect_with_connection(connection: Connection, stop: &AtomicBool) -> Result<Self, String> {
         let queue = connection.new_event_queue();
         connection.display().get_registry(&queue.handle(), ());
         let mut this = Self {
@@ -135,8 +139,8 @@ impl Capture {
         Ok(())
     }
 
-    pub fn validate_sources(&mut self, selected: &[(String, u32)]) -> Result<(), String> {
-        self.sync(&AtomicBool::new(false))?;
+    pub fn validate_sources(&mut self, selected: &[(String, u32)], stop: &AtomicBool) -> Result<(), String> {
+        self.sync(stop)?;
         for (name, generation) in selected {
             if self.generation(name) != Some(*generation) {
                 return Err("A selected display disconnected during approval".into());
@@ -534,6 +538,41 @@ impl Dispatch<window_manager::FereseWindowCaptureManagerV1, ()> for State {
 #[cfg(test)]
 mod tests {
     use super::*;
+
+    #[test]
+    fn discovery_and_final_validation_cancel_on_a_stalled_compositor() {
+        use std::io::Read;
+        use std::os::unix::net::UnixStream;
+
+        for discovery in [true, false] {
+            let (client, mut server) = UnixStream::pair().unwrap();
+            server.set_read_timeout(Some(Duration::from_secs(2))).unwrap();
+            let stop = Arc::new(AtomicBool::new(false));
+            let flag = stop.clone();
+            let worker = std::thread::spawn(move || {
+                let connection = Connection::from_socket(client).unwrap();
+                if discovery {
+                    Capture::connect_with_connection(connection, &flag).err().unwrap()
+                } else {
+                    let queue = connection.new_event_queue();
+                    let mut capture = Capture {
+                        connection,
+                        queue,
+                        state: State::default(),
+                        selected_output: None,
+                    };
+                    capture.validate_sources(&[], &flag).unwrap_err()
+                }
+            });
+            // Wait for the real Wayland request before cancelling its roundtrip.
+            assert!(server.read(&mut [0u8; 64]).unwrap() > 0);
+            let cancelled = Instant::now();
+            stop.store(true, Ordering::SeqCst);
+            assert!(worker.join().unwrap().to_lowercase().contains("cancel"));
+            assert!(cancelled.elapsed() < Duration::from_secs(1));
+        }
+    }
+
     #[test]
     fn capture_orientation_matches_presentation_for_all_transforms() {
         use wl_output::Transform::*;

portal/src/desktop.rs

@@ -235,7 +235,7 @@ pub(crate) async fn ipc(command: &'static str, args: serde_json::Value) -> Resul
         .map_err(|error| error.to_string())?
 }
 
-fn ipc_sync(command: &'static str, args: serde_json::Value) -> Result<serde_json::Value, String> {
+pub(crate) fn ipc_sync(command: &'static str, args: serde_json::Value) -> Result<serde_json::Value, String> {
     let path = std::env::var_os("XDG_RUNTIME_DIR")
         .map(PathBuf::from)
         .ok_or("Missing runtime directory")?
@@ -294,8 +294,16 @@ async fn select_geometry(point: bool) -> Result<Option<String>, String> {
 }
 
 pub(crate) async fn wait_for_surface_removal(pid: u32) -> Result<(), String> {
+    wait_for_surface_removal_using(|| ipc("has-client-surfaces", serde_json::json!({"pid": pid}))).await
+}
+
+pub(crate) async fn wait_for_surface_removal_using<
+    F: std::future::Future<Output = Result<serde_json::Value, String>>,
+>(
+    mut check: impl FnMut() -> F,
+) -> Result<(), String> {
     for _ in 0..100 {
-        if ipc("has-client-surfaces", serde_json::json!({"pid": pid})).await? == false {
+        if check().await? == false {
             return Ok(());
         }
         tokio::time::sleep(Duration::from_millis(10)).await;
@@ -633,7 +641,7 @@ async fn wallpaper(
         return Ok(None);
     }
     let (_, path) = file.keep().map_err(|error| error.to_string())?;
-    if let Err(error) = update_wallpaper(&path, target) {
+    if let Err(error) = update_wallpaper(&path, target).await {
         let _ = std::fs::remove_file(path);
         return Err(error);
     }
@@ -699,44 +707,99 @@ fn prepare_wallpaper(input: &Path, cancel: &Cancel) -> Result<tempfile::NamedTem
     Ok(file)
 }
 
-fn update_wallpaper(image: &Path, target: &str) -> Result<(), String> {
-    let _transaction = crate::backend::CONFIG_TRANSACTION
-        .lock()
-        .map_err(|_| "Configuration transaction failed")?;
-    let path = ferese_config::config_path().ok_or("Missing config path")?;
-    edit_wallpaper(&path, image, target)
+async fn update_wallpaper(image: &Path, target: &str) -> Result<(), String> {
+    let value = crate::bridge::Bridge::connect()?
+        .call("theme-get", serde_json::json!({}))
+        .await?;
+    let snapshot = ferese_ipc::theme::Snapshot::decode(value, ferese_config::families::builtins)?;
+    let image = image.to_owned();
+    let target = target.to_owned();
+    tokio::task::spawn_blocking(move || {
+        let _transaction = crate::backend::CONFIG_TRANSACTION
+            .lock()
+            .map_err(|_| "Configuration transaction failed")?;
+        let path = ferese_config::config_path().ok_or("Missing config path")?;
+        edit_wallpaper(&path, &image, &target, snapshot.theme.appearance)
+    })
+    .await
+    .map_err(|error| error.to_string())?
 }
 
-fn edit_wallpaper(path: &Path, image: &Path, target: &str) -> Result<(), String> {
+fn wallpaper_theme(
+    document: &ferese_config::Document,
+    directory: &Path,
+    appearance: ferese_config::theme::Appearance,
+) -> Result<ferese_config::theme::ResolvedTheme, String> {
+    let mut document = document.clone();
+    let mode = match appearance {
+        ferese_config::theme::Appearance::Light => "light",
+        ferese_config::theme::Appearance::Dark => "dark",
+    };
+    // The compositor knows the active appearance, including system overrides.
+    // Resolve that variant without changing the user's automatic policy.
+    document
+        .set("theme.mode", mode.into())
+        .map_err(|error| error.to_string())?;
+    ferese_config::theme::resolve(&document, directory, jiff::Timestamp::now(), |file| {
+        std::fs::read_to_string(file).map_err(|error| error.to_string())
+    })
+    .map(|candidate| candidate.theme)
+}
+
+fn edit_wallpaper(
+    path: &Path,
+    image: &Path,
+    target: &str,
+    appearance: ferese_config::theme::Appearance,
+) -> Result<(), String> {
+    use ferese_config::theme::Appearance;
+
     let source = match std::fs::read_to_string(path) {
         Ok(source) => source,
         Err(error) if error.kind() == std::io::ErrorKind::NotFound => String::new(),
         Err(error) => return Err(error.to_string()),
     };
     let mut document = ferese_config::Document::parse(&source).map_err(|error| error.to_string())?;
-    if target == "background" && document.get("theme.background.lock_path").is_none() {
-        let old = document
-            .get("theme.background.path")
-            .and_then(serde_json::Value::as_str)
-            .unwrap_or(ferese_config::default_wallpaper())
-            .to_owned();
+    let directory = path.parent().ok_or("Invalid config path")?;
+    let before = wallpaper_theme(&document, directory, appearance)?.tokens.background;
+    let (variant, other) = match appearance {
+        Appearance::Light => ("light", Appearance::Dark),
+        Appearance::Dark => ("dark", Appearance::Light),
+    };
+    let other_before = wallpaper_theme(&document, directory, other)?.tokens.background;
+    let base = format!("theme.{variant}.background");
+    if target == "background" && before.lock_path.is_none() {
+        let old = before
+            .path
+            .unwrap_or_else(|| ferese_config::default_wallpaper_for(appearance).into());
         document
-            .set("theme.background.lock_path", old.into())
+            .set(&format!("{base}.lock_path"), old.to_string_lossy().into_owned().into())
             .map_err(|error| error.to_string())?;
     }
+
     if target != "lockscreen" {
         document
-            .set("theme.background.path", image.to_string_lossy().into_owned().into())
+            .set(&format!("{base}.path"), image.to_string_lossy().into_owned().into())
             .map_err(|error| error.to_string())?;
     }
+
     if target != "background" {
         document
             .set(
-                "theme.background.lock_path",
+                &format!("{base}.lock_path"),
                 image.to_string_lossy().into_owned().into(),
             )
             .map_err(|error| error.to_string())?;
     }
+
+    let after = wallpaper_theme(&document, directory, appearance)?.tokens.background;
+    if (target != "lockscreen" && after.path.as_deref() != Some(image))
+        || (target != "background" && after.lock_path.as_deref() != Some(image))
+        || wallpaper_theme(&document, directory, other)?.tokens.background != other_before
+    {
+        return Err("Wallpaper change did not resolve to the requested image".into());
+    }
+
     let parent = path.parent().ok_or("Invalid config path")?;
     std::fs::create_dir_all(parent).map_err(|error| error.to_string())?;
     let mut temp = tempfile::NamedTempFile::new_in(parent).map_err(|error| error.to_string())?;
@@ -821,27 +884,144 @@ mod tests {
 
     #[test]
     fn wallpaper_targets_preserve_independent_paths_and_other_settings() {
+        use ferese_config::theme::Appearance::{Dark, Light};
+
+        for appearance in [Light, Dark] {
+            let dir = tempfile::tempdir().unwrap();
+            let path = dir.path().join("config.kdl");
+            std::fs::write(
+                &path,
+                "// Keep my settings\ntheme { background { path \"/old.png\"; }; }; commands { terminal \"foot\"; }\n",
+            )
+            .unwrap();
+            let resolve = |mode| {
+                let document = ferese_config::Document::parse(&std::fs::read_to_string(&path).unwrap()).unwrap();
+                wallpaper_theme(&document, dir.path(), mode).unwrap().tokens.background
+            };
+            let other = if appearance == Light { Dark } else { Light };
+            let untouched = resolve(other);
+            edit_wallpaper(&path, Path::new("/desktop.png"), "background", appearance).unwrap();
+            assert_eq!(resolve(appearance).path.as_deref(), Some(Path::new("/desktop.png")));
+            assert_eq!(resolve(appearance).lock_path.as_deref(), Some(Path::new("/old.png")));
+            assert_eq!(resolve(other), untouched);
+            assert!(std::fs::read_to_string(&path).unwrap().contains("Keep my settings"));
+
+            edit_wallpaper(&path, Path::new("/lock.png"), "lockscreen", appearance).unwrap();
+            assert_eq!(resolve(appearance).path.as_deref(), Some(Path::new("/desktop.png")));
+            assert_eq!(resolve(appearance).lock_path.as_deref(), Some(Path::new("/lock.png")));
+            assert_eq!(resolve(other), untouched);
+
+            edit_wallpaper(&path, Path::new("/both.png"), "both", appearance).unwrap();
+            assert_eq!(resolve(appearance).path.as_deref(), Some(Path::new("/both.png")));
+            assert_eq!(resolve(appearance).lock_path.as_deref(), Some(Path::new("/both.png")));
+            assert_eq!(resolve(other), untouched);
+            let document = ferese_config::Document::parse(&std::fs::read_to_string(&path).unwrap()).unwrap();
+            assert_eq!(document.get("theme.background.path").unwrap(), "/old.png");
+            assert_eq!(document.get("commands.terminal").unwrap(), &serde_json::json!(["foot"]));
+        }
+    }
+
+    #[test]
+    fn wallpaper_replaces_active_overrides_without_changing_auto_policy() {
+        use ferese_config::theme::Appearance::{Dark, Light};
+
+        for appearance in [Light, Dark] {
+            let dir = tempfile::tempdir().unwrap();
+            let path = dir.path().join("config.kdl");
+            std::fs::write(
+                &path,
+                r#"theme {
+                mode "auto"
+                schedule { source "system"; }
+                background { path "/shared.png"; lock-path "/shared-lock.png"; }
+                light { background { path "light.png"; lock-path "light-lock.png"; }; }
+                dark { background { path "dark.png"; lock-path "dark-lock.png"; }; }
+            }
+            "#,
+            )
+            .unwrap();
+            let before = ferese_config::Document::parse(&std::fs::read_to_string(&path).unwrap()).unwrap();
+            let other = if appearance == Light { Dark } else { Light };
+            let old_active = wallpaper_theme(&before, dir.path(), appearance)
+                .unwrap()
+                .tokens
+                .background;
+            let old_other = wallpaper_theme(&before, dir.path(), other).unwrap().tokens.background;
+            edit_wallpaper(&path, Path::new("/new.png"), "background", appearance).unwrap();
+            let after = ferese_config::Document::parse(&std::fs::read_to_string(&path).unwrap()).unwrap();
+            let current = wallpaper_theme(&after, dir.path(), appearance)
+                .unwrap()
+                .tokens
+                .background;
+            assert_eq!(current.path.as_deref(), Some(Path::new("/new.png")));
+            assert_eq!(current.lock_path, old_active.lock_path);
+            assert_eq!(
+                wallpaper_theme(&after, dir.path(), other).unwrap().tokens.background,
+                old_other
+            );
+            assert_eq!(after.get("theme.mode").unwrap(), "auto");
+            assert_eq!(after.get("theme.background"), before.get("theme.background"));
+
+            edit_wallpaper(&path, Path::new("/new-lock.png"), "lockscreen", appearance).unwrap();
+            let after = ferese_config::Document::parse(&std::fs::read_to_string(&path).unwrap()).unwrap();
+            let current = wallpaper_theme(&after, dir.path(), appearance)
+                .unwrap()
+                .tokens
+                .background;
+            assert_eq!(current.path.as_deref(), Some(Path::new("/new.png")));
+            assert_eq!(current.lock_path.as_deref(), Some(Path::new("/new-lock.png")));
+            assert_eq!(
+                wallpaper_theme(&after, dir.path(), other).unwrap().tokens.background,
+                old_other
+            );
+        }
+    }
+
+    #[test]
+    fn desktop_only_change_pins_the_effective_variant_lock_wallpaper() {
+        use ferese_config::theme::Appearance::{Dark, Light};
+
         let dir = tempfile::tempdir().unwrap();
         let path = dir.path().join("config.kdl");
         std::fs::write(
             &path,
-            "// Keep my settings\ntheme { background { path \"/old.png\"; }; }; commands { terminal \"foot\"; }\n",
+            r#"theme {
+            background { path "shared.png"; }
+            light { background { path "light.png"; }; }
+            dark { background { path "dark.png"; }; }
+        }
+        "#,
         )
         .unwrap();
-        edit_wallpaper(&path, Path::new("/desktop.png"), "background").unwrap();
-        let document = ferese_config::Document::parse(&std::fs::read_to_string(&path).unwrap()).unwrap();
-        assert_eq!(document.get("theme.background.path").unwrap(), "/desktop.png");
-        assert_eq!(document.get("theme.background.lock_path").unwrap(), "/old.png");
-        assert!(std::fs::read_to_string(&path).unwrap().contains("Keep my settings"));
-        edit_wallpaper(&path, Path::new("/lock.png"), "lockscreen").unwrap();
-        let document = ferese_config::Document::parse(&std::fs::read_to_string(&path).unwrap()).unwrap();
-        assert_eq!(document.get("theme.background.path").unwrap(), "/desktop.png");
-        assert_eq!(document.get("theme.background.lock_path").unwrap(), "/lock.png");
-        edit_wallpaper(&path, Path::new("/both.png"), "both").unwrap();
-        let document = ferese_config::Document::parse(&std::fs::read_to_string(&path).unwrap()).unwrap();
-        assert_eq!(document.get("theme.background.path").unwrap(), "/both.png");
-        assert_eq!(document.get("theme.background.lock_path").unwrap(), "/both.png");
-        assert_eq!(document.get("commands.terminal").unwrap(), &serde_json::json!(["foot"]));
+        let before = ferese_config::Document::parse(&std::fs::read_to_string(&path).unwrap()).unwrap();
+        let previous = wallpaper_theme(&before, dir.path(), Dark).unwrap().tokens.background;
+        assert_eq!(previous.lock_path, None);
+        edit_wallpaper(&path, Path::new("/new.png"), "background", Dark).unwrap();
+        let after = ferese_config::Document::parse(&std::fs::read_to_string(&path).unwrap()).unwrap();
+        let current = wallpaper_theme(&after, dir.path(), Dark).unwrap().tokens.background;
+        assert_eq!(current.lock_path, previous.path);
+        assert_eq!(
+            wallpaper_theme(&after, dir.path(), Light).unwrap().tokens.background,
+            wallpaper_theme(&before, dir.path(), Light).unwrap().tokens.background
+        );
+    }
+
+    #[test]
+    fn wallpaper_validation_failure_preserves_config() {
+        let dir = tempfile::tempdir().unwrap();
+        let path = dir.path().join("config.kdl");
+        let source = "theme { dark { file \"missing.kdl\"; }; }\n";
+        std::fs::write(&path, source).unwrap();
+        assert!(
+            edit_wallpaper(
+                &path,
+                Path::new("/new.png"),
+                "both",
+                ferese_config::theme::Appearance::Dark
+            )
+            .is_err()
+        );
+        assert_eq!(std::fs::read_to_string(&path).unwrap(), source);
     }
 
     #[test]

portal/src/input_capture.rs

@@ -105,6 +105,20 @@ impl InputCapture {
         }
     }
 
+    async fn finish_native_call(
+        &self,
+        connection: &Connection,
+        path: &OwnedObjectPath,
+        bridge: &Bridge,
+        result: Result<serde_json::Value, String>,
+    ) -> zbus::fdo::Result<serde_json::Value> {
+        if bridge.is_closed() {
+            self.end(connection, path.as_str()).await;
+        }
+
+        result.map_err(failed)
+    }
+
     async fn native(session: &Session) -> zbus::fdo::Result<(Bridge, u64)> {
         let state = session.state.lock().await;
         if !state.started || session.cancel.stopped.load(Ordering::SeqCst) {
@@ -127,9 +141,16 @@ impl InputCapture {
         bridge: Bridge,
         id: u64,
     ) {
+        let control = session.state.lock().await.bridge.clone();
+        let Some(control) = control else {
+            self.end(&connection, path.as_str()).await;
+            return;
+        };
+
         loop {
             let response = tokio::select! {
                 _ = session.cancel.wait() => break,
+                _ = control.terminated() => break,
                 result = bridge.capture_watch(id) => result,
             };
             let Ok(response) = response else {
@@ -453,7 +474,9 @@ impl InputCapture {
         if cancel.stopped.load(Ordering::SeqCst) || session.cancel.stopped.load(Ordering::SeqCst) || result.is_none() {
             return Ok((1, Options::new()));
         }
-        let result = result.unwrap().map_err(failed)?;
+        let result = self
+            .finish_native_call(connection, &session_handle, &bridge, result.unwrap())
+            .await?;
         let zones: Vec<(u32, u32, i32, i32)> =
             serde_json::from_value(result["zones"].clone()).map_err(|_| invalid("Invalid native capture zones"))?;
         Ok((
@@ -515,7 +538,9 @@ impl InputCapture {
             self.end(connection, session_handle.as_str()).await;
             return Ok((1, Options::new()));
         }
-        let result = result.unwrap().map_err(failed)?;
+        let result = self
+            .finish_native_call(connection, &session_handle, &bridge, result.unwrap())
+            .await?;
         let failed_barriers: Vec<u32> = serde_json::from_value(result["failed_barriers"].clone())
             .map_err(|_| invalid("Invalid native barrier response"))?;
         Ok((
@@ -580,10 +605,9 @@ impl InputCapture {
             }
             tokio::time::sleep(Duration::from_millis(10)).await;
         }
-        bridge
-            .call("input-capture-enable", json!({"session":id}))
-            .await
-            .map_err(failed)?;
+        let result = bridge.call("input-capture-enable", json!({"session":id})).await;
+        self.finish_native_call(connection, &session_handle, &bridge, result)
+            .await?;
         Ok((0, Options::new()))
     }
 
@@ -597,10 +621,9 @@ impl InputCapture {
     ) -> zbus::fdo::Result<(u32, Options)> {
         let session = self.session(connection, &header, &session_handle, &app_id).await?;
         let (bridge, id) = Self::native(&session).await?;
-        bridge
-            .call("input-capture-disable", json!({"session":id}))
-            .await
-            .map_err(failed)?;
+        let result = bridge.call("input-capture-disable", json!({"session":id})).await;
+        self.finish_native_call(connection, &session_handle, &bridge, result)
+            .await?;
         Ok((0, Options::new()))
     }
 
@@ -633,13 +656,14 @@ impl InputCapture {
                 Ok(position)
             })
             .transpose()?;
-        bridge
+        let result = bridge
             .call(
                 "input-capture-release",
                 json!({"session":id,"activation_id":activation,"cursor_position":position}),
             )
-            .await
-            .map_err(failed)?;
+            .await;
+        self.finish_native_call(connection, &session_handle, &bridge, result)
+            .await?;
         Ok((0, Options::new()))
     }
 
@@ -705,6 +729,74 @@ impl SessionObject {
 mod tests {
     use super::*;
 
+    #[tokio::test]
+    async fn broken_control_transport_ends_capture_but_valid_rejection_does_not() {
+        use std::process::Stdio;
+        use tokio::io::{AsyncBufReadExt, BufReader};
+
+        let mut bus = tokio::process::Command::new("dbus-daemon")
+            .args(["--session", "--nofork", "--print-address=1"])
+            .stdout(Stdio::piped())
+            .kill_on_drop(true)
+            .spawn()
+            .unwrap();
+        let mut address = String::new();
+        BufReader::new(bus.stdout.take().unwrap())
+            .read_line(&mut address)
+            .await
+            .unwrap();
+        let connection = zbus::connection::Builder::address(address.trim())
+            .unwrap()
+            .build()
+            .await
+            .unwrap();
+
+        for transport_failure in [false, true] {
+            let capture = InputCapture::default();
+            let path: OwnedObjectPath = "/org/freedesktop/portal/desktop/session/test/capture"
+                .try_into()
+                .unwrap();
+            let (control, mut peer) = Bridge::test_pair();
+            let (watch, _peer) = Bridge::test_pair();
+            let session = Arc::new(Session {
+                owner: connection.unique_name().unwrap().to_string(),
+                app: "test".into(),
+                cancel: Arc::default(),
+                state: Mutex::new(State {
+                    started: true,
+                    bridge: Some(control.clone()),
+                    watch: Some(watch.clone()),
+                    ..State::default()
+                }),
+            });
+            capture.sessions.lock().await.insert(path.to_string(), session.clone());
+            let response = std::thread::spawn(move || {
+                let request: ferese_ipc::Request = ferese_ipc::read_frame(&mut peer).unwrap();
+                if !transport_failure {
+                    ferese_ipc::write_frame(&mut peer, &ferese_ipc::Response::error(request.id, "denied", "Denied"))
+                        .unwrap();
+                }
+            });
+            let result = control.call("input-capture-disable", json!({})).await;
+            assert!(
+                capture
+                    .finish_native_call(&connection, &path, &control, result)
+                    .await
+                    .is_err()
+            );
+            response.join().unwrap();
+            assert_eq!(session.cancel.stopped.load(Ordering::SeqCst), transport_failure);
+            assert_eq!(watch.is_closed(), transport_failure);
+            assert_eq!(
+                capture.sessions.lock().await.contains_key(path.as_str()),
+                !transport_failure
+            );
+            capture.end(&connection, path.as_str()).await;
+        }
+        bus.kill().await.unwrap();
+        bus.wait().await.unwrap();
+    }
+
     #[test]
     fn requested_capabilities_are_required_and_supported_subset_is_returned() {
         assert!(capabilities(&Options::new()).is_err());