💩 zb: Workaround for xdg-dbus-proxy's monotonic serial requirement

zbus/src/abstractions/async_lock.rs

@@ -2,3 +2,37 @@
 pub(crate) use async_lock::{Mutex, RwLock, RwLockReadGuard, RwLockWriteGuard};
 #[cfg(feature = "tokio")]
 pub(crate) use tokio::sync::{Mutex, RwLock, RwLockReadGuard, RwLockWriteGuard};
+
+/// An abstraction over async semaphore API.
+#[cfg(not(feature = "tokio"))]
+pub(crate) struct Semaphore(async_lock::Semaphore);
+#[cfg(feature = "tokio")]
+pub(crate) struct Semaphore(tokio::sync::Semaphore);
+
+impl Semaphore {
+    pub const fn new(permits: usize) -> Self {
+        #[cfg(not(feature = "tokio"))]
+        let semaphore = async_lock::Semaphore::new(permits);
+        #[cfg(feature = "tokio")]
+        let semaphore = tokio::sync::Semaphore::const_new(permits);
+
+        Self(semaphore)
+    }
+
+    pub async fn acquire(&self) -> SemaphorePermit<'_> {
+        #[cfg(not(feature = "tokio"))]
+        {
+            self.0.acquire().await
+        }
+        #[cfg(feature = "tokio")]
+        {
+            // SAFETY: Since we never explicitly close the sempaphore, `acquire` can't fail.
+            self.0.acquire().await.unwrap()
+        }
+    }
+}
+
+#[cfg(not(feature = "tokio"))]
+pub(crate) type SemaphorePermit<'a> = async_lock::SemaphoreGuard<'a>;
+#[cfg(feature = "tokio")]
+pub(crate) type SemaphorePermit<'a> = tokio::sync::SemaphorePermit<'a>;

zbus/src/connection/handshake/client.rs

@@ -4,7 +4,7 @@ use tracing::{debug, instrument, trace, warn};
 
 use sha1::{Digest, Sha1};
 
-use crate::{conn::socket::ReadHalf, names::OwnedUniqueName, Message};
+use crate::{conn::socket::ReadHalf, is_flatpak, names::OwnedUniqueName, Message};
 
 use super::{
     random_ascii, sasl_auth_id, AuthMechanism, Authenticated, BoxedSplit, Command, Common, Cookie,
@@ -317,7 +317,3 @@ async fn receive_hello_response(
         m => Err(Error::Handshake(format!("Unexpected messgage `{m:?}`"))),
     }
 }
-
-fn is_flatpak() -> bool {
-    std::env::var("FLATPAK_ID").is_ok()
-}

zbus/src/connection/mod.rs

@@ -21,9 +21,10 @@ use futures_core::Future;
 use futures_util::StreamExt;
 
 use crate::{
-    async_lock::Mutex,
+    async_lock::{Mutex, Semaphore, SemaphorePermit},
     blocking,
     fdo::{self, ConnectionCredentials, RequestNameFlags, RequestNameReply},
+    is_flatpak,
     message::{Flags, Message, Type},
     proxy::CacheProperties,
     DBusError, Error, Executor, MatchRule, MessageStream, ObjectServer, OwnedGuid, OwnedMatchRule,
@@ -351,6 +352,8 @@ impl Connection {
         M::Error: Into<Error>,
         B: serde::ser::Serialize + zvariant::DynamicType,
     {
+        let _permit = acquire_serial_num_semaphore().await;
+
         let mut builder = Message::method(path, method_name)?;
         if let Some(sender) = self.unique_name() {
             builder = builder.sender(sender)?
@@ -404,6 +407,8 @@ impl Connection {
         M::Error: Into<Error>,
         B: serde::ser::Serialize + zvariant::DynamicType,
     {
+        let _permit = acquire_serial_num_semaphore().await;
+
         let mut b = Message::signal(path, interface, signal_name)?;
         if let Some(sender) = self.unique_name() {
             b = b.sender(sender)?;
@@ -424,6 +429,8 @@ impl Connection {
     where
         B: serde::ser::Serialize + zvariant::DynamicType,
     {
+        let _permit = acquire_serial_num_semaphore().await;
+
         let mut b = Message::method_reply(call)?;
         if let Some(sender) = self.unique_name() {
             b = b.sender(sender)?;
@@ -442,6 +449,8 @@ impl Connection {
         E: TryInto<ErrorName<'e>>,
         E::Error: Into<Error>,
     {
+        let _permit = acquire_serial_num_semaphore().await;
+
         let mut b = Message::method_error(call, error_name)?;
         if let Some(sender) = self.unique_name() {
             b = b.sender(sender)?;
@@ -459,6 +468,8 @@ impl Connection {
         call: &zbus::message::Header<'_>,
         err: impl DBusError,
     ) -> Result<()> {
+        let _permit = acquire_serial_num_semaphore().await;
+
         let m = err.create_reply(call)?;
         self.send(&m).await
     }
@@ -1273,6 +1284,20 @@ enum NameStatus {
     Queued(#[allow(unused)] Task<()>),
 }
 
+static SERIAL_NUM_SEMAPHORE: Semaphore = Semaphore::new(1);
+
+// Make message creation and sending an atomic operation, using an async
+// semaphore if flatpak portal is detected to workaround an xdg-dbus-proxy issue:
+//
+// https://github.com/flatpak/xdg-dbus-proxy/issues/46
+async fn acquire_serial_num_semaphore() -> Option<SemaphorePermit<'static>> {
+    if is_flatpak() {
+        Some(SERIAL_NUM_SEMAPHORE.acquire().await)
+    } else {
+        None
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

zbus/src/utils.rs

@@ -48,3 +48,8 @@ pub fn block_on<F: std::future::Future>(future: F) -> F::Output {
     });
     runtime.block_on(future)
 }
+
+// If we're running inside a Flatpak sandbox.
+pub(crate) fn is_flatpak() -> bool {
+    std::env::var("FLATPAK_ID").is_ok()
+}