diff --git a/shell/ferese-shell/src/status.rs b/shell/ferese-shell/src/status.rs index f8987f3..c91fb85 100644 --- a/shell/ferese-shell/src/status.rs +++ b/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() } diff --git a/shell/ferese-shell/src/status/bluetooth.rs b/shell/ferese-shell/src/status/bluetooth.rs index 2086c06..5aacd8c 100644 --- a/shell/ferese-shell/src/status/bluetooth.rs +++ b/shell/ferese-shell/src/status/bluetooth.rs @@ -13,15 +13,47 @@ pub(super) fn read(connection: &Connection) -> zbus::Result> { )? .body() .deserialize()?; - Ok(snapshot(&objects)) + let adapters = objects + .iter() + .filter_map(|(path, interfaces)| { + interfaces + .get("org.bluez.Adapter1") + .map(|properties| (path, properties)) + }) + .collect::>(); + // 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 { - // 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 { 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] diff --git a/shell/ferese-shell/src/status/dbus_cache.rs b/shell/ferese-shell/src/status/dbus_cache.rs new file mode 100644 index 0000000..127f262 --- /dev/null +++ b/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, + alive: Arc, + _stop: oneshot::Sender<()>, +} + +impl Signals { + fn new(connection: &Connection, service: &str) -> zbus::Result { + // 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 { + bus: StatusBus, + service: &'static str, + signals: Option, + revision: u64, + action: u64, + value: Option, + retry_at: Instant, +} + +impl Cache { + 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 { + 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); + #[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> { + 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" + ); + } +}