Freeze buf fix

.github/.gitignore

@@ -0,0 +1,2 @@
+# Ignore the prompts directory
+prompts/

Cargo.toml

@@ -10,7 +10,7 @@ rust-version = "1.88"
 
 [dependencies]
 serde = { version = "1", features = ["derive"] }
-tokio = { version = "1", features = [] }
+tokio = { version = "1", features = ["time", "net", "io-util"] }
 
 i18n-embed = { version = "0.16", features = [
     "fluent-system",

src/app.rs

@@ -374,17 +374,23 @@ impl<Db: DbTrait + 'static> cosmic::Application for AppState<Db> {
                         error!("can't insert data: {e}");
                     }
                 }
-                clipboard::ClipboardMessage::Error(e) => {
-                    error!("clipboard: {e}");
-
-                    self.clipboard_state = if let ClipboardError::Watch(ref e) = e
-                        && let clipboard_watcher::Error::MissingProtocol { name, .. } = **e
-                        && name == "zwlr_data_control_manager_v1"
-                    {
-                        ClipboardState::Error(ErrorState::MissingDataControlProtocol)
-                    } else {
-                        ClipboardState::Error(ErrorState::Other(e.to_string()))
-                    };
+                clipboard::ClipboardMessage::ErrorRecoverable(e) => {
+                    warn!("clipboard recoverable error: {e}");
+                    // For recoverable errors, we keep the subscription running
+                    // and just log the error. The clipboard state remains unchanged.
+                }
+                clipboard::ClipboardMessage::ErrorFatal(e) => {
+                    error!("clipboard fatal error: {e}");
+
+                    let ClipboardError::Watch(ref inner) = e;
+                    self.clipboard_state =
+                        if let clipboard_watcher::Error::MissingProtocol { name, .. } = **inner
+                            && name == "zwlr_data_control_manager_v1"
+                        {
+                            ClipboardState::Error(ErrorState::MissingDataControlProtocol)
+                        } else {
+                            ClipboardState::Error(ErrorState::Other(e.to_string()))
+                        };
                 }
                 clipboard::ClipboardMessage::EmptyKeyboard => {
                     if let Some(data) = self.db.get(0) {

src/clipboard.rs

@@ -1,5 +1,5 @@
 use std::{
-    io::Read,
+    os::fd::{FromRawFd, IntoRawFd, OwnedFd},
     sync::{
         Arc,
         atomic::{self},
@@ -8,8 +8,11 @@ use std::{
 
 use cosmic::iced::{futures::SinkExt, stream::channel};
 use futures::Stream;
+use futures::future::join_all;
 use itertools::Itertools;
+use tokio::io::AsyncReadExt;
 use tokio::sync::mpsc;
+use tokio::time::{Duration, timeout};
 
 use crate::{clipboard_watcher, config::PRIVATE_MODE, db::MimeDataMap};
 
@@ -20,7 +23,10 @@ pub enum ClipboardMessage {
     /// Means that the source was closed, or the compurer just started
     /// This means the clipboard manager must become the source, by providing the last entry
     EmptyKeyboard,
-    Error(ClipboardError),
+    /// Recoverable error - can potentially retry
+    ErrorRecoverable(ClipboardError),
+    /// Fatal error - requires intervention or configuration change
+    ErrorFatal(ClipboardError),
 }
 
 #[derive(Debug, Clone, thiserror::Error)]
@@ -29,6 +35,22 @@ pub enum ClipboardError {
     Watch(Arc<clipboard_watcher::Error>),
 }
 
+impl ClipboardError {
+    /// Returns true if this error is potentially recoverable (transient)
+    pub fn is_recoverable(&self) -> bool {
+        match self {
+            ClipboardError::Watch(e) => matches!(
+                **e,
+                clipboard_watcher::Error::ClipboardEmpty
+                    | clipboard_watcher::Error::WaylandCommunication(_)
+                    | clipboard_watcher::Error::WaylandFlush(_)
+                    | clipboard_watcher::Error::OfferNotFound
+                    | clipboard_watcher::Error::RegistryInvalidId
+            ),
+        }
+    }
+}
+
 enum WatchRes<I> {
     Some(I),
     None,
@@ -48,24 +70,39 @@ pub fn sub() -> impl Stream<Item = ClipboardMessage> {
                         {
                             Ok(res) => {
                                 if !PRIVATE_MODE.load(atomic::Ordering::Relaxed) {
-                                    tx.blocking_send(WatchRes::Some(res)).unwrap();
+                                    if tx.blocking_send(WatchRes::Some(res)).is_err() {
+                                        debug!(
+                                            "clipboard channel receiver dropped, exiting watcher loop"
+                                        );
+                                        break;
+                                    }
                                 } else {
                                     info!("private mode")
                                 }
                             }
                             Err(e) => match e {
                                 clipboard_watcher::Error::ClipboardEmpty => {
-                                    tx.blocking_send(WatchRes::None).unwrap();
+                                    if tx.blocking_send(WatchRes::None).is_err() {
+                                        debug!(
+                                            "clipboard channel receiver dropped, exiting watcher loop"
+                                        );
+                                        break;
+                                    }
                                 }
                                 _ => {
-                                    tx.blocking_send(WatchRes::Err(e)).unwrap();
+                                    // Try to send error, but don't panic if channel is closed
+                                    let _ = tx.blocking_send(WatchRes::Err(e));
                                     break;
                                 }
                             },
                         }
                     }
                 });
-                output.send(ClipboardMessage::Connected).await.unwrap();
+
+                if output.send(ClipboardMessage::Connected).await.is_err() {
+                    warn!("clipboard output channel closed during connect");
+                    return;
+                }
 
                 let mut i = 0;
                 loop {
@@ -77,23 +114,51 @@ pub fn sub() -> impl Stream<Item = ClipboardMessage> {
                         Some(WatchRes::Some(res)) => {
                             let mut data = MimeDataMap::new();
 
-                            for (mime_type, mut pipe) in res {
-                                let mut contents = Vec::new();
+                            let read_futures = res.into_iter().map(|(mime_type, pipe)| async move {
+                                let raw_fd = pipe.into_raw_fd();
+                                let owned_fd = unsafe { OwnedFd::from_raw_fd(raw_fd) };
+
+                                let mut async_pipe =
+                                    match tokio::net::unix::pipe::Receiver::from_owned_fd(owned_fd)
+                                    {
+                                        Ok(receiver) => receiver,
+                                        Err(e) => {
+                                            warn!(
+                                                "failed to create async pipe receiver: {mime_type} {e}"
+                                            );
+                                            return None;
+                                        }
+                                    };
 
-                                match pipe.read_to_end(&mut contents) {
-                                    Ok(len) => {
+                                let mut contents = Vec::new();
+                                match timeout(
+                                    Duration::from_millis(500),
+                                    async_pipe.read_to_end(&mut contents),
+                                )
+                                .await
+                                {
+                                    Ok(Ok(len)) => {
                                         if len == 0 {
                                             debug!("data is empty: {mime_type}");
+                                            None
                                         } else {
-                                            data.insert(mime_type, contents);
+                                            Some((mime_type, contents))
                                         }
                                     }
-                                    Err(e) => {
-                                        warn!(
-                                            "read error on external pipe clipboard: {mime_type} {e}"
-                                        );
+                                    Ok(Err(e)) => {
+                                        warn!("read error on external pipe clipboard: {mime_type} {e}");
+                                        None
+                                    }
+                                    Err(_) => {
+                                        warn!("read timeout (500ms): {mime_type}");
+                                        None
                                     }
                                 }
+                            });
+
+                            let read_results = join_all(read_futures).await;
+                            for (mime_type, contents) in read_results.into_iter().flatten() {
+                                data.insert(mime_type, contents);
                             }
 
                             if !data.is_empty() {
@@ -103,37 +168,50 @@ pub fn sub() -> impl Stream<Item = ClipboardMessage> {
                                     .collect_vec();
 
                                 debug!("send mime types to db: {mimes:?}");
-                                output.send(ClipboardMessage::Data(data)).await.unwrap();
+                                if output.send(ClipboardMessage::Data(data)).await.is_err() {
+                                    warn!("clipboard output channel closed");
+                                    return;
+                                }
                             }
                         }
 
                         Some(WatchRes::None) => {
                             debug!("empty keyboard");
-                            output.send(ClipboardMessage::EmptyKeyboard).await.unwrap();
+                            if output.send(ClipboardMessage::EmptyKeyboard).await.is_err() {
+                                warn!("clipboard output channel closed");
+                                return;
+                            }
                         }
                         Some(WatchRes::Err(e)) => {
-                            output
-                                .send(ClipboardMessage::Error(ClipboardError::Watch(e.into())))
-                                .await
-                                .unwrap();
-                            std::future::pending::<()>().await;
+                            let error = ClipboardError::Watch(e.into());
+                            let message = if error.is_recoverable() {
+                                ClipboardMessage::ErrorRecoverable(error)
+                            } else {
+                                ClipboardMessage::ErrorFatal(error)
+                            };
+                            if output.send(message).await.is_err() {
+                                warn!("clipboard output channel closed during error report");
+                            }
+                            return;
                         }
                         None => {
-                            std::future::pending::<()>().await;
+                            debug!("clipboard watcher channel closed");
+                            return;
                         }
                     }
                 }
             }
 
             Err(e) => {
-                // todo: how to cancel properly?
-                // https://github.com/pop-os/cosmic-files/blob/d96d48995d49e17f01903ca4d89839eb4a1b1104/src/app.rs#L1704
-                output
-                    .send(ClipboardMessage::Error(ClipboardError::Watch(e.into())))
-                    .await
-                    .unwrap();
-
-                std::future::pending::<()>().await;
+                let error = ClipboardError::Watch(e.into());
+                let message = if error.is_recoverable() {
+                    ClipboardMessage::ErrorRecoverable(error)
+                } else {
+                    ClipboardMessage::ErrorFatal(error)
+                };
+                if output.send(message).await.is_err() {
+                    warn!("clipboard output channel closed during init error report");
+                }
             }
         };
     })