Cache network and Bluetooth status until D-Bus changes

shell/ferese-shell/src/status.rs

@@ -1,5 +1,6 @@
 //! Bounded, off-UI-thread adapters. A missing service is `None`, never a fake state.
 mod bluetooth;
+mod dbus_cache;
 mod network;
 
 use std::{
@@ -145,6 +146,8 @@ impl Service {
         thread::spawn(move || {
             let mut system_bus = StatusBus::new(true);
             let mut session_bus = StatusBus::new(false);
+            let mut network = dbus_cache::Cache::new("org.freedesktop.NetworkManager");
+            let mut bluetooth = dbus_cache::Cache::new("org.bluez");
             loop {
                 // A poll overlapping a write is discarded below. Wait for the
                 // write to finish before retrying instead of launching commands
@@ -152,7 +155,9 @@ impl Service {
                 let Some(before) = wait_for_poll(&polling) else {
                     break;
                 };
-                let snapshot = poll(&mut system_bus, &mut session_bus);
+                let mut snapshot = poll(&mut system_bus, &mut session_bus);
+                snapshot.network = network.read(before, network::read);
+                snapshot.bluetooth = bluetooth.read(before, bluetooth::read);
                 let state = polling.0.lock().unwrap();
                 if state.3 {
                     break;
@@ -337,8 +342,8 @@ impl StatusBus {
 
 fn poll(system_bus: &mut StatusBus, session_bus: &mut StatusBus) -> Snapshot {
     Snapshot {
-        network: system_bus.query(network::read).flatten(),
-        bluetooth: system_bus.query(bluetooth::read).flatten(),
+        network: None,
+        bluetooth: None,
         audio: audio(),
         battery: battery(),
         brightness: brightness(),
@@ -540,7 +545,7 @@ mod tests {
         pub(super) fn connect(&self) -> zbus::blocking::Connection {
             zbus::blocking::connection::Builder::address(self.address.as_str())
                 .unwrap()
-                .method_timeout(Duration::from_millis(200))
+                .method_timeout(Duration::from_secs(2))
                 .build()
                 .unwrap()
         }

shell/ferese-shell/src/status/bluetooth.rs

@@ -13,15 +13,47 @@ pub(super) fn read(connection: &Connection) -> zbus::Result<Option<Bluetooth>> {
         )?
         .body()
         .deserialize()?;
-    Ok(snapshot(&objects))
+    let adapters = objects
+        .iter()
+        .filter_map(|(path, interfaces)| {
+            interfaces
+                .get("org.bluez.Adapter1")
+                .map(|properties| (path, properties))
+        })
+        .collect::<Vec<_>>();
+    // BlueZ has no default-controller property. With several controllers, use
+    // bluetoothctl's own selection so the displayed toggle and CLI action
+    // address the same controller. The signal cache avoids idle CLI polling.
+    let selected = if adapters.len() > 1 {
+        let Some(info) = super::run("bluetoothctl", &["show"]).ok() else {
+            return Ok(None);
+        };
+        let Some(address) = info.lines().find_map(|line| {
+            line.strip_prefix("Controller ")
+                .and_then(|line| line.split_whitespace().next())
+        }) else {
+            return Ok(None);
+        };
+        let Some((path, _)) = adapters.iter().find(|(_, properties)| {
+            properties
+                .get("Address")
+                .and_then(|value| <&str>::try_from(value).ok())
+                == Some(address)
+        }) else {
+            return Ok(None);
+        };
+        Some(path.as_str())
+    } else {
+        None
+    };
+    Ok(snapshot(&objects, selected))
 }
 
-fn snapshot(objects: &ManagedObjects) -> Option<Bluetooth> {
-    // bluetoothctl's fresh process uses its first controller. Prefer hci0 and
-    // keep the selection stable when the object-manager dictionary is reordered.
+fn snapshot(objects: &ManagedObjects, selected: Option<&str>) -> Option<Bluetooth> {
     let (adapter, properties) = objects
         .iter()
         .filter_map(|(path, interfaces)| interfaces.get("org.bluez.Adapter1").map(|p| (path, p)))
+        .filter(|(path, _)| selected.is_none_or(|selected| selected == path.as_str()))
         .min_by(|(a, _), (b, _)| a.as_str().cmp(b.as_str()))?;
     let enabled = bool::try_from(properties.get("Powered")?).ok()?;
     let mut devices = Vec::new();
@@ -105,7 +137,8 @@ mod tests {
     #[test]
     fn selects_one_adapter_and_only_its_connected_devices() {
         let mut objects = graph();
-        let value = snapshot(&objects).unwrap();
+        assert!(!snapshot(&objects, Some("/org/bluez/hci1")).unwrap().enabled);
+        let value = snapshot(&objects, Some("/org/bluez/hci0")).unwrap();
         assert!(value.enabled);
         assert_eq!(value.devices, ["Headphones"]);
         let adapter = objects
@@ -114,11 +147,11 @@ mod tests {
             .get_mut("org.bluez.Adapter1")
             .unwrap();
         adapter.insert("Powered".into(), false.into());
-        let value = snapshot(&objects).unwrap();
+        let value = snapshot(&objects, None).unwrap();
         assert!(!value.enabled && value.devices.is_empty());
         adapter_removed(&mut objects);
-        assert!(!snapshot(&objects).unwrap().enabled);
-        assert!(snapshot(&ManagedObjects::new()).is_none());
+        assert!(!snapshot(&objects, None).unwrap().enabled);
+        assert!(snapshot(&ManagedObjects::new(), None).is_none());
     }
     fn adapter_removed(objects: &mut ManagedObjects) {
         objects.retain(|path, _| !path.as_str().starts_with("/org/bluez/hci0"));
@@ -131,13 +164,15 @@ mod tests {
                 adapter.remove("Powered");
             }
         }
-        assert!(snapshot(&objects).is_none());
+        assert!(snapshot(&objects, None).is_none());
     }
     struct Manager;
     #[zbus::interface(name = "org.freedesktop.DBus.ObjectManager")]
     impl Manager {
         fn get_managed_objects(&self) -> ManagedObjects {
-            graph()
+            let mut objects = graph();
+            objects.retain(|path, _| !path.as_str().starts_with("/org/bluez/hci1"));
+            objects
         }
     }
     #[test]

shell/ferese-shell/src/status/dbus_cache.rs

@@ -0,0 +1,273 @@
+//! Keep native snapshots until the owning service signals a change.
+use super::StatusBus;
+use cosmic::iced::futures::{FutureExt, StreamExt, channel::oneshot};
+use std::sync::{
+    Arc,
+    atomic::{AtomicBool, AtomicU64, Ordering},
+};
+use std::time::{Duration, Instant};
+use zbus::blocking::{Connection, MessageIterator};
+
+struct Signals {
+    revision: Arc<AtomicU64>,
+    alive: Arc<AtomicBool>,
+    _stop: oneshot::Sender<()>,
+}
+
+impl Signals {
+    fn new(connection: &Connection, service: &str) -> zbus::Result<Self> {
+        // Install both rules before the snapshot, including when the service is
+        // currently absent. Owner changes invalidate paths after a restart.
+        let changes = MessageIterator::for_match_rule(
+            zbus::MatchRule::builder()
+                .msg_type(zbus::message::Type::Signal)
+                .sender(service)?
+                .build(),
+            connection,
+            Some(64),
+        )?;
+        let owners = MessageIterator::for_match_rule(
+            zbus::MatchRule::builder()
+                .msg_type(zbus::message::Type::Signal)
+                .sender("org.freedesktop.DBus")?
+                .interface("org.freedesktop.DBus")?
+                .member("NameOwnerChanged")?
+                .add_arg(service)?
+                .build(),
+            connection,
+            Some(8),
+        )?;
+        let revision = Arc::new(AtomicU64::new(1));
+        let alive = Arc::new(AtomicBool::new(true));
+        let changed = revision.clone();
+        let running = alive.clone();
+        let (stop, stopped) = oneshot::channel::<()>();
+        std::thread::spawn(move || {
+            // Enter a runtime for zbus stream cleanup, including cancellation.
+            let runtime = tokio::runtime::Builder::new_current_thread()
+                .enable_all()
+                .build();
+            if let Ok(runtime) = runtime {
+                // Dropping the blocking wrappers enters zbus' runtime, so do
+                // this before entering our async runtime.
+                let changes = changes.into_inner();
+                let owners = owners.into_inner();
+                runtime.block_on(async move {
+                    let mut events = cosmic::iced::futures::stream::select(changes, owners);
+                    let mut stopped = stopped.fuse();
+                    loop {
+                        cosmic::iced::futures::select! {
+                            _ = stopped => break,
+                            event = events.next() => {
+                                changed.fetch_add(1, Ordering::Release);
+                                if !matches!(event, Some(Ok(_))) { break; }
+                            }
+                        }
+                    }
+                    let (changes, owners) = events.into_inner();
+                    zbus::AsyncDrop::async_drop(changes).await;
+                    zbus::AsyncDrop::async_drop(owners).await;
+                });
+            }
+            running.store(false, Ordering::Release);
+        });
+        Ok(Self {
+            revision,
+            alive,
+            _stop: stop,
+        })
+    }
+}
+
+pub(super) struct Cache<T> {
+    bus: StatusBus,
+    service: &'static str,
+    signals: Option<Signals>,
+    revision: u64,
+    action: u64,
+    value: Option<T>,
+    retry_at: Instant,
+}
+
+impl<T: Clone> Cache<T> {
+    pub(super) fn new(service: &'static str) -> Self {
+        Self {
+            bus: StatusBus::new(true),
+            service,
+            signals: None,
+            revision: 0,
+            action: u64::MAX,
+            value: None,
+            retry_at: Instant::now(),
+        }
+    }
+
+    pub(super) fn read(
+        &mut self,
+        action: u64,
+        read: fn(&Connection) -> zbus::Result<Option<T>>,
+    ) -> Option<T> {
+        if self
+            .signals
+            .as_ref()
+            .is_some_and(|s| !s.alive.load(Ordering::Acquire))
+        {
+            self.signals = None;
+            self.bus.connection = None;
+            self.value = None;
+        }
+        if self.signals.is_none() {
+            self.signals = self
+                .bus
+                .query(|connection| Signals::new(connection, self.service));
+            self.revision = 0;
+        }
+        // If subscription setup fails, read fresh rather than trusting a cache
+        // with no invalidation. Retry setup on the next regular status poll.
+        let revision = self
+            .signals
+            .as_ref()
+            .map(|s| s.revision.load(Ordering::Acquire));
+        if revision.is_none()
+            || revision != Some(self.revision)
+            || action != self.action
+            || (self.value.is_none() && Instant::now() >= self.retry_at)
+        {
+            // Keep signal subscriptions across service errors. A dead bus is
+            // detected by the listener; a transient query failure retries later.
+            self.value = self
+                .bus
+                .query(|connection| Ok(connection.clone()))
+                .and_then(|connection| read(&connection).ok().flatten());
+            self.retry_at = Instant::now() + Duration::from_secs(5);
+            self.action = action;
+            self.revision = revision.unwrap_or(0);
+        }
+        self.value.clone()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::time::{Duration, Instant};
+    struct Counter(Arc<AtomicU64>);
+    #[zbus::interface(name = "org.ferese.StatusCounter")]
+    impl Counter {
+        fn read(&self) -> u64 {
+            self.0.fetch_add(1, Ordering::Relaxed) + 1
+        }
+    }
+    fn read_counter(connection: &Connection) -> zbus::Result<Option<u64>> {
+        connection
+            .call_method(
+                Some("org.ferese.StatusCounter"),
+                "/test",
+                Some("org.ferese.StatusCounter"),
+                "Read",
+                &(),
+            )?
+            .body()
+            .deserialize()
+            .map(Some)
+    }
+    #[test]
+    #[ignore = "requires dbus-daemon; uses a private test bus"]
+    fn cache_skips_idle_reads_but_refreshes_after_actions_signals_and_failures() {
+        let bus = crate::status::tests::TestBus::new();
+        let count = Arc::new(AtomicU64::new(0));
+        let server = zbus::blocking::connection::Builder::address(bus.address.as_str())
+            .unwrap()
+            .name("org.ferese.StatusCounter")
+            .unwrap()
+            .serve_at("/test", Counter(count.clone()))
+            .unwrap()
+            .build()
+            .unwrap();
+        let mut cache = Cache::new("org.ferese.StatusCounter");
+        cache.bus.connection = Some(bus.connect());
+        assert_eq!(cache.read(0, read_counter), Some(1));
+        for _ in 0..10 {
+            assert_eq!(cache.read(0, read_counter), Some(1));
+        }
+        assert_eq!(count.load(Ordering::Relaxed), 1);
+        assert_eq!(cache.read(1, read_counter), Some(2));
+        server
+            .emit_signal(
+                None::<&str>,
+                "/test",
+                "org.ferese.StatusCounter",
+                "Changed",
+                &(),
+            )
+            .unwrap();
+        let deadline = Instant::now() + Duration::from_secs(2);
+        while cache.read(1, read_counter) != Some(3) {
+            assert!(Instant::now() < deadline, "signal did not invalidate cache");
+            std::thread::sleep(Duration::from_millis(1));
+        }
+        server.release_name("org.ferese.StatusCounter").unwrap();
+        let deadline = Instant::now() + Duration::from_secs(2);
+        while cache.read(1, read_counter).is_some() {
+            assert!(Instant::now() < deadline, "owner loss kept stale data");
+            std::thread::sleep(Duration::from_millis(1));
+        }
+        server.request_name("org.ferese.StatusCounter").unwrap();
+        let deadline = Instant::now() + Duration::from_secs(2);
+        while cache.read(1, read_counter).is_none() {
+            assert!(Instant::now() < deadline, "owner return did not recover");
+            std::thread::sleep(Duration::from_millis(1));
+        }
+        // A transient method error may have no subsequent signal. It must
+        // clear the snapshot immediately and retry after the bounded backoff.
+        assert!(
+            cache
+                .read(2, |_| Err(zbus::Error::Failure("temporary".into())))
+                .is_none()
+        );
+        assert!(cache.read(2, read_counter).is_none());
+        cache.retry_at = Instant::now();
+        assert!(cache.read(2, read_counter).is_some());
+    }
+
+    #[test]
+    #[ignore = "requires dbus-daemon; uses a private test bus"]
+    fn signals_invalidate_and_owner_loss_is_observed() {
+        let bus = crate::status::tests::TestBus::new();
+        let client = bus.connect();
+        let server = bus.connect();
+        server.request_name("org.ferese.StatusTest").unwrap();
+        let signals = Signals::new(&client, "org.ferese.StatusTest").unwrap();
+        let wait = |before| {
+            let deadline = Instant::now() + Duration::from_secs(2);
+            while signals.revision.load(Ordering::Acquire) == before && Instant::now() < deadline {
+                std::thread::sleep(Duration::from_millis(1));
+            }
+            assert!(signals.revision.load(Ordering::Acquire) > before);
+        };
+        let before = signals.revision.load(Ordering::Acquire);
+        server
+            .emit_signal(
+                None::<&str>,
+                "/test",
+                "org.ferese.StatusTest",
+                "Changed",
+                &(),
+            )
+            .unwrap();
+        wait(before);
+        let before = signals.revision.load(Ordering::Acquire);
+        server.release_name("org.ferese.StatusTest").unwrap();
+        wait(before);
+        let alive = signals.alive.clone();
+        drop(signals);
+        let deadline = Instant::now() + Duration::from_secs(2);
+        while alive.load(Ordering::Acquire) && Instant::now() < deadline {
+            std::thread::sleep(Duration::from_millis(1));
+        }
+        assert!(
+            !alive.load(Ordering::Acquire),
+            "dropping the service must stop its listener"
+        );
+    }
+}