diff --git a/crates/openlogi-agent-core/src/orchestrator.rs b/crates/openlogi-agent-core/src/orchestrator.rs index eff033c2..090c8f04 100644 --- a/crates/openlogi-agent-core/src/orchestrator.rs +++ b/crates/openlogi-agent-core/src/orchestrator.rs @@ -11,13 +11,13 @@ //! (still valid) values — exactly the GUI's "window never opened" behaviour. use std::collections::{BTreeMap, HashSet}; -use std::sync::atomic::{AtomicI32, Ordering}; +use std::sync::atomic::{AtomicI32, AtomicU64, Ordering}; use std::sync::{Arc, RwLock}; use openlogi_core::config::{Config, ScrollResolution}; use openlogi_core::device::{Capabilities, DeviceInventory}; use openlogi_hid::{CaptureChannel, DeviceRoute}; -use tracing::warn; +use tracing::{debug, warn}; use crate::DpiCycleState; use crate::bindings::{bindings_for, gesture_bindings_for, oshook_gestures_for}; @@ -58,6 +58,10 @@ pub struct SharedRuntime { pub dpi_cycle: Arc>, pub thumbwheel_sensitivity: Arc, pub capture_channel: CaptureChannel, + /// Incremented when the selected device reconnects or the system wakes, so + /// the gesture watcher re-arms volatile HID++ control diversion even when + /// the receiver route itself never changed. + pub capture_rearm_generation: Arc, /// Exclusive receiver access shared by HID++ capture and pairing. Capture /// and pairing must never open the same receiver HID node concurrently. pub receiver_access: ReceiverAccess, @@ -111,6 +115,7 @@ impl Orchestrator { config.app_settings.thumbwheel_sensitivity, )), capture_channel: Arc::new(RwLock::new(None)), + capture_rearm_generation: Arc::new(AtomicU64::new(0)), receiver_access: ReceiverAccess::default(), }; let orch = Self { @@ -205,6 +210,9 @@ impl Orchestrator { // (offline→online), or — via the // flag — a system wake where none of those are observable. let reapply_all = std::mem::take(&mut self.reapply_all_next_refresh); + let next_current = pick_current(&devices, self.config.selected_device()); + let rearm_capture = + selected_needs_capture_rearm(&self.devices, &devices, next_current, reapply_all); let followup = std::mem::take(&mut self.reapply_followup); let (targets, next_followup) = plan_reapply(&self.devices, &devices, &followup, reapply_all); @@ -218,15 +226,23 @@ impl Orchestrator { || a.route != b.route || a.capabilities != b.capabilities }); - if !changed { + if changed { + self.devices = devices; + self.current = next_current; + self.rebuild(); + } else { // Same set and routes — but keep the fresh `online` flags, or a // device that woke this tick would read as a transition forever. self.devices = devices; - return; } - self.devices = devices; - self.current = pick_current(&self.devices, self.config.selected_device()); - self.rebuild(); + if rearm_capture { + let generation = self + .shared + .capture_rearm_generation + .fetch_add(1, Ordering::Relaxed) + .wrapping_add(1); + debug!(generation, "selected device requires capture re-arm"); + } } /// Force a volatile-settings re-apply for every online device on the next @@ -466,6 +482,18 @@ fn reapply_targets(prev: &[AgentDevice], next: &[AgentDevice], reapply_all: bool .collect() } +/// Whether this refresh invalidated the selected device's volatile control +/// diversion. Receiver routes stay connected while a paired mouse sleeps, so +/// route equality alone cannot tell the capture watcher to re-arm on wake. +fn selected_needs_capture_rearm( + prev: &[AgentDevice], + next: &[AgentDevice], + selected: usize, + reapply_all: bool, +) -> bool { + reapply_targets(prev, next, reapply_all).contains(&selected) +} + /// Plan this refresh's volatile-settings writes: the [`reapply_targets`] set /// plus one confirming re-apply for devices first sighted last refresh, and /// the follow-up keys to confirm next refresh. @@ -520,7 +548,7 @@ fn write_value(lock: &RwLock, value: T, name: &str) { mod tests { use super::{ AgentDevice, InventoryHealth, Orchestrator, build_devices, configured_wheel_mode, - plan_reapply, reapply_targets, + plan_reapply, reapply_targets, selected_needs_capture_rearm, }; use openlogi_core::config::{Config, ScrollResolution}; use openlogi_core::device::{ @@ -674,6 +702,29 @@ mod tests { assert_eq!(reapply_targets(&prev, &next, true), vec![0]); } + #[test] + fn selected_receiver_reconnect_requests_capture_rearm() { + let prev = [dev("selected", 1, false), dev("other", 2, true)]; + let next = [dev("selected", 1, true), dev("other", 2, true)]; + + assert!(selected_needs_capture_rearm(&prev, &next, 0, false)); + assert!(!selected_needs_capture_rearm(&prev, &next, 1, false)); + } + + #[test] + fn system_wake_requests_capture_rearm_for_selected_online_device() { + let devices = [dev("selected", 1, true), dev("other", 2, true)]; + + assert!(selected_needs_capture_rearm(&devices, &devices, 0, true)); + } + + #[test] + fn steady_inventory_does_not_cycle_capture() { + let devices = [dev("selected", 1, true)]; + + assert!(!selected_needs_capture_rearm(&devices, &devices, 0, false)); + } + #[test] fn plan_reapply_confirms_a_first_sighting_once() { use std::collections::HashSet; diff --git a/crates/openlogi-agent-core/src/watchers/gesture.rs b/crates/openlogi-agent-core/src/watchers/gesture.rs index 80659682..4aae1bec 100644 --- a/crates/openlogi-agent-core/src/watchers/gesture.rs +++ b/crates/openlogi-agent-core/src/watchers/gesture.rs @@ -19,14 +19,14 @@ //! way regardless. use std::collections::BTreeMap; -use std::sync::atomic::{AtomicI32, Ordering}; +use std::sync::atomic::{AtomicI32, AtomicU64, Ordering}; use std::sync::{Arc, RwLock}; use std::thread; use std::time::{Duration, Instant}; use openlogi_core::binding::{Action, ButtonId, GestureDirection, default_binding}; use openlogi_core::config::DEFAULT_THUMBWHEEL_SENSITIVITY; -use openlogi_hid::{CaptureChannel, CapturedInput, DeviceRoute, run_capture_session}; +use openlogi_hid::{CaptureChannel, CaptureStop, CapturedInput, DeviceRoute, run_capture_session}; use tokio::sync::{mpsc, oneshot}; use tracing::{debug, warn}; @@ -80,6 +80,7 @@ pub fn spawn( dpi_cycle: Arc>, capture_channel: CaptureChannel, thumbwheel_sensitivity: ThumbwheelSensitivity, + capture_rearm_generation: Arc, receiver_access: ReceiverAccess, ) { thread::spawn(move || { @@ -99,6 +100,7 @@ pub fn spawn( dpi_cycle, capture_channel, thumbwheel_sensitivity, + capture_rearm_generation, receiver_access, )); }); @@ -141,6 +143,27 @@ fn should_rearm(done_epoch: u64, live_epoch: u64, has_target: bool) -> bool { done_epoch == live_epoch && has_target } +#[derive(Debug, Clone, PartialEq, Eq)] +struct CaptureTarget { + route: DeviceRoute, + capture_thumbwheel: bool, + divert_gesture_button: bool, + rearm_generation: u64, +} + +/// A generation-only restart on the same route follows a device reconnect or +/// system wake. Its old firmware state is already gone, so restoring it would +/// only delay (or permanently block) the replacement session. +fn stop_for_transition(current: &CaptureTarget, next: Option<&CaptureTarget>) -> CaptureStop { + if next.is_some_and(|next| { + next.route == current.route && next.rearm_generation != current.rearm_generation + }) { + CaptureStop::Abandon + } else { + CaptureStop::Restore + } +} + /// Keep one capture session alive for the active device, restarting it when the /// device or the thumb-wheel arming changes, and dispatch incoming inputs. Runs /// for the lifetime of the process. @@ -150,12 +173,12 @@ async fn manage( dpi_cycle: Arc>, capture_channel: CaptureChannel, thumbwheel_sensitivity: ThumbwheelSensitivity, + capture_rearm_generation: Arc, receiver_access: ReceiverAccess, ) { let (tx, mut rx) = mpsc::unbounded_channel::(); - // (route, capture_thumbwheel, divert_gesture_button) - let mut current: Option<(DeviceRoute, bool, bool)> = None; - let mut stop: Option> = None; + let mut current: Option = None; + let mut stop: Option> = None; let mut ticker = tokio::time::interval(TARGET_POLL); let mut accumulators = WheelAccumulators::default(); // Capture sessions run as detached tasks, so an unexpected exit (a transient @@ -197,33 +220,39 @@ async fn manage( // thread the full config in. Re-evaluated each tick, so a // ReloadConfig owner change restarts the session accordingly. let divert_gesture = gesture_bindings.read().is_ok_and(|g| !g.is_empty()); - target.map(|t| { - ( - t, - thumbwheel_armed(&hook_maps, sensitivity), - divert_gesture, - ) + let rearm_generation = capture_rearm_generation.load(Ordering::Relaxed); + target.map(|route| CaptureTarget { + route, + capture_thumbwheel: thumbwheel_armed(&hook_maps, sensitivity), + divert_gesture_button: divert_gesture, + rearm_generation, }) }; if want == current { continue; } + debug!(?current, ?want, "capture target state changed"); // Target or thumb-wheel arming changed (or first tick): stop the // old session and start one for the new state. Sending on the // oneshot lets the old session restore the diverted controls. if let Some(stop) = stop.take() { - let _ = stop.send(()); + let reason = current + .as_ref() + .map_or(CaptureStop::Restore, |current| { + stop_for_transition(current, want.as_ref()) + }); + let _ = stop.send(reason); } if current.is_some() { current = None; continue; } - if let Some((route, capture_thumbwheel, divert_gesture_button)) = want { + if let Some(target) = want { let Some(receiver_lease) = receiver_access.try_acquire_for_capture() else { current = None; continue; }; - current = Some((route.clone(), capture_thumbwheel, divert_gesture_button)); + current = Some(target.clone()); let (stop_tx, stop_rx) = oneshot::channel(); let sink = tx.clone(); let slot = Arc::clone(&capture_channel); @@ -233,9 +262,9 @@ async fn manage( tokio::spawn(async move { let _receiver_lease = receiver_lease; if let Err(e) = run_capture_session( - route, - capture_thumbwheel, - divert_gesture_button, + target.route, + target.capture_thumbwheel, + target.divert_gesture_button, sink, stop_rx, slot, @@ -445,8 +474,58 @@ fn advance( #[cfg(test)] mod tests { + use openlogi_hid::{CaptureStop, DeviceRoute}; + use super::*; + fn capture_target(route: DeviceRoute, generation: u64) -> CaptureTarget { + CaptureTarget { + route, + capture_thumbwheel: false, + divert_gesture_button: true, + rearm_generation: generation, + } + } + + #[test] + fn reconnect_restart_abandons_stale_state_on_the_same_route() { + let route = DeviceRoute::Unifying { + receiver_uid: "receiver".to_string(), + slot: 1, + }; + let current = capture_target(route.clone(), 3); + let next = capture_target(route, 4); + + assert_eq!( + stop_for_transition(¤t, Some(&next)), + CaptureStop::Abandon + ); + } + + #[test] + fn ordinary_target_changes_restore_old_controls() { + let current = capture_target( + DeviceRoute::Unifying { + receiver_uid: "receiver".to_string(), + slot: 1, + }, + 3, + ); + let next = capture_target( + DeviceRoute::Direct { + vendor_id: 0x046d, + product_id: 0xb023, + }, + 4, + ); + + assert_eq!( + stop_for_transition(¤t, Some(&next)), + CaptureStop::Restore + ); + assert_eq!(stop_for_transition(¤t, None), CaptureStop::Restore); + } + #[test] fn multiplier_is_unity_at_default_sensitivity() { assert!((scroll_multiplier(DEFAULT_THUMBWHEEL_SENSITIVITY) - 1.0).abs() < f32::EPSILON); diff --git a/crates/openlogi-agent/src/main.rs b/crates/openlogi-agent/src/main.rs index 6ce40c45..3da24d67 100644 --- a/crates/openlogi-agent/src/main.rs +++ b/crates/openlogi-agent/src/main.rs @@ -168,6 +168,7 @@ async fn run(config: Config) { shared.dpi_cycle.clone(), shared.capture_channel.clone(), shared.thumbwheel_sensitivity.clone(), + shared.capture_rearm_generation.clone(), shared.receiver_access.clone(), ); diff --git a/crates/openlogi-agent/src/pairing.rs b/crates/openlogi-agent/src/pairing.rs index ef365590..1387577a 100644 --- a/crates/openlogi-agent/src/pairing.rs +++ b/crates/openlogi-agent/src/pairing.rs @@ -270,6 +270,7 @@ mod tests { dpi_cycle: Arc::new(RwLock::new(DpiCycleState::default())), thumbwheel_sensitivity: Arc::new(0.into()), capture_channel: Arc::new(RwLock::new(None)), + capture_rearm_generation: Arc::new(0.into()), receiver_access: ReceiverAccess::default(), } } diff --git a/crates/openlogi-hid/src/gesture.rs b/crates/openlogi-hid/src/gesture.rs index 1783494b..2a7d06e4 100644 --- a/crates/openlogi-hid/src/gesture.rs +++ b/crates/openlogi-hid/src/gesture.rs @@ -16,6 +16,7 @@ //! is therefore only diverted when its click is actually bound. use std::sync::{Arc, Mutex, PoisonError, RwLock}; +use std::time::Duration; use hidpp::{channel::HidppChannel, device::Device, protocol::v20}; use openlogi_core::binding::{ButtonId, GestureDirection, SwipeAccumulator}; @@ -34,6 +35,16 @@ use crate::write::SharedChannel; /// whenever no session is connected. pub type CaptureChannel = Arc>>; +/// How an established capture session should release its diverted controls. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum CaptureStop { + /// Restore each diverted control to its firmware default before exiting. + Restore, + /// Exit without writes because a reconnect already discarded the device's + /// volatile diversion state. + Abandon, +} + /// One input captured from the active device. #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] pub enum CapturedInput { @@ -64,6 +75,36 @@ pub enum GestureError { /// A HID++ feature call returned an error; inner string carries context. #[error("HID++ protocol error: {0}")] Hidpp(String), + /// An established HID channel disconnected while capture was active. + #[error("HID channel disconnected")] + ChannelDisconnected, +} + +const CAPTURE_HEALTH_POLL: Duration = Duration::from_secs(1); + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum CaptureExit { + Stopped(CaptureStop), + Disconnected, +} + +async fn wait_for_capture_exit( + chan: &HidppChannel, + mut shutdown: oneshot::Receiver, + poll_period: Duration, +) -> CaptureExit { + loop { + tokio::select! { + stop = &mut shutdown => { + return CaptureExit::Stopped(stop.unwrap_or(CaptureStop::Restore)); + } + () = tokio::time::sleep(poll_period) => { + if !chan.is_connected() { + return CaptureExit::Disconnected; + } + } + } + } } /// Movement + button state accumulated across messages. Lives behind a `Mutex` @@ -90,14 +131,17 @@ struct CaptureAccum { /// /// Opens and holds one HID++ channel, diverts whichever of those controls the /// device exposes, and listens. Returns once `shutdown` fires (or its sender is -/// dropped), after restoring every diverted control. Setup errors are returned; -/// failures to restore on the way out are logged, not propagated. +/// dropped), or when the established channel disconnects. A normal shutdown +/// restores every diverted control; [`CaptureStop::Abandon`] and a disconnect +/// skip restoration because the device has already reset its volatile mappings. +/// Setup errors are returned; failures to restore on the way out are logged, +/// not propagated. pub async fn run_capture_session( route: DeviceRoute, capture_thumbwheel: bool, divert_gesture_button: bool, sink: mpsc::UnboundedSender, - shutdown: oneshot::Receiver<()>, + shutdown: oneshot::Receiver, channel_slot: CaptureChannel, ) -> Result<(), GestureError> { let chan = open_route_channel(&route) @@ -159,15 +203,30 @@ pub async fn run_capture_session( thumbwheel = armed.thumb.is_some(), "control capture active" ); - let _ = shutdown.await; + let exit = wait_for_capture_exit(&chan, shutdown, CAPTURE_HEALTH_POLL).await; drop(listener); if let Ok(mut slot) = channel_slot.write() { *slot = None; } - armed.disarm().await; - debug!(index = device_index, "control capture stopped"); - Ok(()) + match exit { + CaptureExit::Stopped(CaptureStop::Restore) => { + armed.disarm().await; + debug!(index = device_index, "control capture stopped"); + Ok(()) + } + CaptureExit::Stopped(CaptureStop::Abandon) => { + debug!( + index = device_index, + "control capture abandoned after reconnect" + ); + Ok(()) + } + CaptureExit::Disconnected => { + debug!(index = device_index, "control capture channel disconnected"); + Err(GestureError::ChannelDisconnected) + } + } } /// The set of controls a session has diverted, kept so they can be handed back diff --git a/crates/openlogi-hid/src/gesture/tests.rs b/crates/openlogi-hid/src/gesture/tests.rs index 035f2cda..4eb5ed25 100644 --- a/crates/openlogi-hid/src/gesture/tests.rs +++ b/crates/openlogi-hid/src/gesture/tests.rs @@ -1,4 +1,99 @@ use super::*; +use std::error::Error; +use std::sync::atomic::{AtomicBool, Ordering}; + +use hidpp::channel::{HidppChannel, RawHidChannel}; + +struct LivenessRawChannel { + connected: Arc, +} + +#[hidpp::async_trait] +impl RawHidChannel for LivenessRawChannel { + fn vendor_id(&self) -> u16 { + 0x046d + } + + fn product_id(&self) -> u16 { + 0xb023 + } + + async fn write_report(&self, src: &[u8]) -> Result> { + Ok(src.len()) + } + + async fn read_report(&self, _buf: &mut [u8]) -> Result> { + std::future::pending().await + } + + fn is_connected(&self) -> bool { + self.connected.load(Ordering::Acquire) + } + + fn supports_short_long_hidpp(&self) -> Option<(bool, bool)> { + Some((false, true)) + } + + async fn get_report_descriptor( + &self, + _buf: &mut [u8], + ) -> Result> { + unreachable!("mock declares HID++ support") + } +} + +#[tokio::test] +async fn established_capture_exits_when_its_channel_disconnects() { + let connected = Arc::new(AtomicBool::new(true)); + let channel = HidppChannel::from_raw_channel(LivenessRawChannel { + connected: Arc::clone(&connected), + }) + .await + .unwrap_or_else(|e| panic!("mock HID++ channel should open: {e}")); + let (_shutdown_tx, shutdown_rx) = oneshot::channel(); + + connected.store(false, Ordering::Release); + let exit = tokio::time::timeout( + Duration::from_millis(50), + wait_for_capture_exit(&channel, shutdown_rx, Duration::from_millis(1)), + ) + .await + .unwrap_or_else(|_| panic!("capture did not notice the disconnected channel")); + + assert_eq!(exit, CaptureExit::Disconnected); +} + +#[tokio::test] +async fn established_capture_honors_shutdown_while_connected() { + let connected = Arc::new(AtomicBool::new(true)); + let channel = HidppChannel::from_raw_channel(LivenessRawChannel { connected }) + .await + .unwrap_or_else(|e| panic!("mock HID++ channel should open: {e}")); + let (shutdown_tx, shutdown_rx) = oneshot::channel(); + shutdown_tx + .send(CaptureStop::Restore) + .unwrap_or_else(|_| panic!("capture shutdown receiver should still be live")); + + let exit = wait_for_capture_exit(&channel, shutdown_rx, Duration::from_millis(1)).await; + + assert_eq!(exit, CaptureExit::Stopped(CaptureStop::Restore)); +} + +#[tokio::test] +async fn established_capture_can_abandon_stale_firmware_state() { + let connected = Arc::new(AtomicBool::new(true)); + let channel = HidppChannel::from_raw_channel(LivenessRawChannel { connected }) + .await + .unwrap_or_else(|e| panic!("mock HID++ channel should open: {e}")); + let (shutdown_tx, shutdown_rx) = oneshot::channel(); + shutdown_tx + .send(CaptureStop::Abandon) + .unwrap_or_else(|_| panic!("capture shutdown receiver should still be live")); + + let exit = wait_for_capture_exit(&channel, shutdown_rx, Duration::from_millis(1)).await; + + assert_eq!(exit, CaptureExit::Stopped(CaptureStop::Abandon)); +} fn press() -> RawControlEvent { RawControlEvent::DivertedButtons([reprog_controls::GESTURE_BUTTON_CID, 0, 0, 0]) diff --git a/crates/openlogi-hid/src/inventory.rs b/crates/openlogi-hid/src/inventory.rs index a95294e4..837c5dfd 100644 --- a/crates/openlogi-hid/src/inventory.rs +++ b/crates/openlogi-hid/src/inventory.rs @@ -2,6 +2,7 @@ use std::{ collections::{HashMap, HashSet}, + hash::Hash, sync::Arc, time::Duration, }; @@ -203,6 +204,50 @@ const ONESHOT_ATTEMPTS: u8 = 4; /// asleep device, so a short pause lets the next attempt read it cleanly. const ONESHOT_RETRY_DELAY: Duration = Duration::from_millis(300); +/// Nodes that remain valid for this tick: everything the OS enumerated plus +/// cached channels whose open transport still reports a live connection. +fn retained_nodes( + enumerated: &HashSet, + cached_channels: impl IntoIterator, +) -> HashSet +where + K: Clone + Eq + Hash, +{ + let mut retained = enumerated.clone(); + retained.extend( + cached_channels + .into_iter() + .filter_map(|(node, connected)| connected.then_some(node)), + ); + retained +} + +/// Add cached channels omitted by this OS enumeration while their open +/// transport still reports a live connection. +fn append_live_cached_channels( + nodes: &mut HashSet, + channels: &HashMap, + active: &mut Vec<(async_hid::DeviceInfo, Arc)>, +) { + let retained = retained_nodes( + nodes, + channels + .iter() + .map(|(node, open)| (node.clone(), open.channel.is_connected())), + ); + for node in retained.difference(nodes) { + if let Some(open) = channels.get(node) { + debug!( + ?node, + name = %open.info.name, + "OS enumeration omitted a live HID node; probing cached channel" + ); + active.push((open.info.clone(), Arc::clone(&open.channel))); + } + } + *nodes = retained; +} + impl Enumerator { /// One enumeration pass, reusing the cache from prior passes. Probes every /// HID candidate concurrently (so one asleep node that burns the whole @@ -270,12 +315,14 @@ impl Enumerator { } } } - // Drop channels for nodes that vanished this tick. A node missing from - // the enumeration is a real disconnect (the IOHIDManager device set is - // authoritative, unlike a HID++ probe timeout), so close the device and - // join its read thread now instead of leaving a dead channel behind; a - // reconnect re-opens under a fresh node id. The ledger forgets vanished - // nodes for the same reason — a true disconnect must not be replayed. + // IOHIDManager can temporarily omit a Bluetooth device's vendor HID++ + // collection while its already-open handle and ordinary mouse link are + // still live. Keep probing that cached channel instead of turning one + // incomplete OS snapshot into an offline device and stopping capture. + append_live_cached_channels(&mut seen_nodes, &self.channels, &mut active); + + // A missing node whose transport confirms disconnection is dropped; + // forget its ledger snapshot so a real disconnect is not replayed. self.channels.retain(|node, _| seen_nodes.contains(node)); self.ledger.retain_nodes(&seen_nodes); diff --git a/crates/openlogi-hid/src/inventory/probe.rs b/crates/openlogi-hid/src/inventory/probe.rs index 8e2b717d..439e4cc1 100644 --- a/crates/openlogi-hid/src/inventory/probe.rs +++ b/crates/openlogi-hid/src/inventory/probe.rs @@ -3,6 +3,7 @@ use std::{collections::HashMap, sync::Arc}; use futures_concurrency::future::Join as _; use hidpp::{ channel::HidppChannel, + device::Device, receiver::{ self, Receiver, bolt::{ @@ -575,39 +576,69 @@ async fn probe_unifying_slot( // online or not, and the crate's `event.online` reads the wrong notification // byte (payload[1] bit6, always set here — wire-verified `04 62 69 40`), so // neither tells us if the device is actually reachable on this receiver. - // We therefore always attempt the probe (passing `true`) and treat the - // feature walk succeeding as the real liveness signal below — a device that - // moved to Bluetooth answers `DeviceNotFound` and surfaces as offline. + // A cache hit must therefore still do one live round-trip: otherwise cached + // capabilities keep an absent device "online" forever and its reconnect is + // invisible to the agent's volatile-state re-apply/capture re-arm path. let probe_result = timeout( UNIFYING_SLOT_PROBE, - probe_or_reuse(channel, slot, Some(id.clone()), cached, true, tick), + probe_unifying_features(channel, slot, &id, cached, tick), ) .await; - let (probe, outcome) = if let Ok(r) = probe_result { + let (probe, outcome, online) = if let Ok(r) = probe_result { r } else { debug!(slot, budget = ?UNIFYING_SLOT_PROBE, "Unifying slot probe timed out; using cached data if available"); let probe = cached.map_or_else(ProbedFeatures::default, |c| c.probe.clone()); - (probe, CacheOutcome::Seen(id)) + (probe, CacheOutcome::Seen(id), false) }; - let device = PairedDevice { + let device = assemble_unifying_device(slot, codename, event.wpid, register_kind, probe, online); + Some((device, outcome)) +} + +/// Return cached immutable features together with a fresh reachability result. +/// +/// A successful full probe ([`CacheOutcome::Fresh`]) confirms liveness on a +/// cache miss/stale entry. A fresh cached entry normally refreshes its battery, +/// whose successful response ([`CacheOutcome::Update`]) is the liveness check. +/// A failed battery refresh, or a device without that feature, gets a root ping +/// before being treated as offline. +pub(super) async fn probe_unifying_features( + channel: &Arc, + slot: u8, + id: &CacheKey, + cached: Option<&Cached>, + tick: u64, +) -> (ProbedFeatures, CacheOutcome, bool) { + let (probe, outcome) = + probe_or_reuse(channel, slot, Some(id.clone()), cached, true, tick).await; + let online = if matches!(outcome, CacheOutcome::Fresh(..) | CacheOutcome::Update(..)) { + true + } else { + Device::new(Arc::clone(channel), slot).await.is_ok() + }; + (probe, outcome, online) +} + +pub(super) fn assemble_unifying_device( + slot: u8, + codename: Option, + wpid: u16, + register_kind: DeviceKind, + probe: ProbedFeatures, + online: bool, +) -> PairedDevice { + PairedDevice { slot, codename, - wpid: Some(event.wpid), + wpid: Some(wpid), kind: resolve_device_kind(probe.kind, register_kind), - // Reachable on this receiver iff the feature walk got through this tick. - // Caveat: a GUI cache hit can serve stale capabilities for up to - // REFRESH_TICKS after the device leaves for Bluetooth, briefly showing it - // online; self-heals on the next forced re-probe. Add a per-tick liveness - // ping if that window ever matters. - online: probe.capabilities.is_some(), + online, battery: probe.battery, model_info: probe.model_info, capabilities: probe.capabilities, - }; - Some((device, outcome)) + } } /// Reads a Unifying paired device's name. Unifying stores names at diff --git a/crates/openlogi-hid/src/inventory/tests.rs b/crates/openlogi-hid/src/inventory/tests.rs index 607cf9b1..a56c2785 100644 --- a/crates/openlogi-hid/src/inventory/tests.rs +++ b/crates/openlogi-hid/src/inventory/tests.rs @@ -1,10 +1,17 @@ -use std::collections::HashSet; +use std::{collections::HashSet, error::Error, io, sync::Arc}; -use openlogi_core::device::{DeviceInventory, DeviceKind, PairedDevice, ReceiverInfo}; +use hidpp::channel::{HidppChannel, RawHidChannel}; +use openlogi_core::device::{ + Capabilities, DeviceInventory, DeviceKind, PairedDevice, ReceiverInfo, +}; +use tokio::sync::{Mutex, mpsc}; use super::cache::{CACHE_MISS_GRACE, CacheKey, CacheOutcome, Cached, REFRESH_TICKS, is_stale}; -use super::probe::{NodeProbe, assemble_bolt_probe, parse_codename_unifying}; -use super::{Enumerator, ONESHOT_ATTEMPTS, one_shot_should_stop}; +use super::probe::{ + NodeProbe, assemble_bolt_probe, assemble_unifying_device, parse_codename_unifying, + probe_unifying_features, +}; +use super::{Enumerator, ONESHOT_ATTEMPTS, one_shot_should_stop, retained_nodes}; use crate::inventory::features::ProbedFeatures; fn cache_entry(probed_tick: u64) -> Cached { @@ -57,6 +64,16 @@ fn being_seen_resets_the_miss_counter() { ); } +#[test] +fn live_cached_channel_survives_a_transient_enumeration_gap() { + let enumerated = HashSet::from([1]); + let cached_channels = [(1, true), (2, true), (3, false)]; + + let retained = retained_nodes(&enumerated, cached_channels); + + assert_eq!(retained, HashSet::from([1, 2])); +} + #[test] fn cached_probe_is_reused_until_refresh_ticks() { let cached = Cached { @@ -75,6 +92,127 @@ fn cached_probe_is_reused_until_refresh_ticks() { ); } +#[test] +fn unifying_cached_features_do_not_override_current_liveness() { + let probe = ProbedFeatures { + capabilities: Some(Capabilities::default()), + ..ProbedFeatures::default() + }; + + let device = assemble_unifying_device( + 1, + Some("cached mouse".to_string()), + 0x4082, + DeviceKind::Mouse, + probe, + false, + ); + + assert!( + !device.online, + "the current liveness probe is authoritative" + ); + assert!( + device.capabilities.is_some(), + "cached immutable features remain available while offline" + ); +} + +struct BatteryErrorPingChannel { + incoming_tx: mpsc::UnboundedSender>, + incoming_rx: Mutex>>, +} + +impl BatteryErrorPingChannel { + fn new() -> Self { + let (incoming_tx, incoming_rx) = mpsc::unbounded_channel(); + Self { + incoming_tx, + incoming_rx: Mutex::new(incoming_rx), + } + } +} + +#[hidpp::async_trait] +impl RawHidChannel for BatteryErrorPingChannel { + fn vendor_id(&self) -> u16 { + 0x046d + } + + fn product_id(&self) -> u16 { + 0xc52b + } + + async fn write_report(&self, src: &[u8]) -> Result> { + let mut response = src.to_vec(); + if src.get(2).copied() != Some(0) { + if response.len() < 7 { + return Err(Box::new(io::Error::other("short mock HID++ report"))); + } + response[2] = 0xff; + response[3] = src[2]; + response[4] = src[3]; + response[5] = 0x08; + response[6] = 0; + } + self.incoming_tx.send(response).map_err(|_| { + Box::new(io::Error::other("mock HID++ response receiver closed")) + as Box + })?; + Ok(src.len()) + } + + async fn read_report(&self, buf: &mut [u8]) -> Result> { + let Some(report) = self.incoming_rx.lock().await.recv().await else { + return Err(Box::new(io::Error::other( + "mock HID++ response sender closed", + ))); + }; + let len = report.len().min(buf.len()); + buf[..len].copy_from_slice(&report[..len]); + Ok(len) + } + + fn supports_short_long_hidpp(&self) -> Option<(bool, bool)> { + Some((true, true)) + } + + async fn get_report_descriptor( + &self, + _buf: &mut [u8], + ) -> Result> { + unreachable!("mock declares HID++ support") + } +} + +#[tokio::test] +async fn unifying_battery_failure_uses_root_ping_before_marking_offline() { + let channel = Arc::new( + HidppChannel::from_raw_channel(BatteryErrorPingChannel::new()) + .await + .unwrap_or_else(|e| panic!("mock HID++ channel should open: {e}")), + ); + let id = CacheKey::UnifyingSlot { + receiver_uid: "receiver".to_string(), + slot: 1, + }; + let cached = Cached { + probe: ProbedFeatures { + capabilities: Some(Capabilities::default()), + ..ProbedFeatures::default() + }, + battery_index: Some(4), + probed_tick: 10, + }; + + let (_, _, online) = probe_unifying_features(&channel, 1, &id, Some(&cached), 11).await; + + assert!( + online, + "a failed battery refresh is inconclusive when the root ping still answers" + ); +} + fn inventory(slots: &[u8]) -> Vec { vec![DeviceInventory { receiver: ReceiverInfo { diff --git a/crates/openlogi-hid/src/lib.rs b/crates/openlogi-hid/src/lib.rs index b6cf1547..b9642f26 100644 --- a/crates/openlogi-hid/src/lib.rs +++ b/crates/openlogi-hid/src/lib.rs @@ -29,7 +29,7 @@ pub mod smartshift; pub mod thumbwheel; pub mod write; -pub use gesture::{CaptureChannel, CapturedInput, GestureError, run_capture_session}; +pub use gesture::{CaptureChannel, CaptureStop, CapturedInput, GestureError, run_capture_session}; pub use hires_wheel::{ ScrollReportingTarget, ScrollResolution, ScrollWheelMode, get_scroll_wheel_mode, get_scroll_wheel_mode_on, set_scroll_inversion, set_scroll_inversion_on, set_scroll_resolution, diff --git a/crates/openlogi-hid/src/transport.rs b/crates/openlogi-hid/src/transport.rs index 7dcc4048..308827e6 100644 --- a/crates/openlogi-hid/src/transport.rs +++ b/crates/openlogi-hid/src/transport.rs @@ -10,6 +10,8 @@ #[cfg(not(target_os = "windows"))] use std::error::Error; +#[cfg(not(target_os = "windows"))] +use std::sync::atomic::AtomicBool; use std::sync::atomic::{AtomicU16, Ordering}; use std::sync::{Arc, LazyLock}; @@ -349,6 +351,7 @@ pub(crate) struct AsyncHidChannel { reader: Mutex, writer: Mutex, info: DeviceInfo, + connected: AtomicBool, /// Whether the device exposes only the long HID++ report (a BLE-direct /// peripheral on macOS). Reported via `supports_short_long_hidpp` so the /// `hidpp` channel up-converts outgoing short messages to long. @@ -367,9 +370,16 @@ impl AsyncHidChannel { reader: Mutex::new(reader), writer: Mutex::new(writer), info, + connected: AtomicBool::new(true), long_only, } } + + fn mark_disconnected(&self) { + if self.connected.swap(false, Ordering::AcqRel) { + debug!(name = %self.info.name, "HID channel disconnected"); + } + } } #[cfg(not(target_os = "windows"))] @@ -385,8 +395,15 @@ impl RawHidChannel for AsyncHidChannel { async fn write_report(&self, src: &[u8]) -> Result> { let mut w = self.writer.lock().await; - w.write_output_report(src).await?; - Ok(src.len()) + match w.write_output_report(src).await { + Ok(()) => Ok(src.len()), + Err(e) => { + if matches!(e, async_hid::HidError::Disconnected) { + self.mark_disconnected(); + } + Err(e.into()) + } + } } async fn read_report(&self, buf: &mut [u8]) -> Result> { @@ -403,11 +420,18 @@ impl RawHidChannel for AsyncHidChannel { // until the inventory watcher evicts the channel), so park instead. // The contract guarantees every caller races this future against // the channel's close signal, which tears the read down on drop. - Err(async_hid::HidError::Disconnected) => std::future::pending().await, + Err(async_hid::HidError::Disconnected) => { + self.mark_disconnected(); + std::future::pending().await + } Err(e) => Err(e.into()), } } + fn is_connected(&self) -> bool { + self.connected.load(Ordering::Acquire) + } + fn supports_short_long_hidpp(&self) -> Option<(bool, bool)> { // USB / receiver collections carry both reports; BLE-direct collections // are long-only (no short report on macOS), where the `hidpp` channel diff --git a/crates/openlogi-hidpp/src/channel.rs b/crates/openlogi-hidpp/src/channel.rs index a8535f6d..f2aef3ab 100755 --- a/crates/openlogi-hidpp/src/channel.rs +++ b/crates/openlogi-hidpp/src/channel.rs @@ -96,6 +96,15 @@ pub trait RawHidChannel: Sync + Send + 'static { /// must do the same and must not await `read_report` bare. async fn read_report(&self, buf: &mut [u8]) -> Result>; + /// Whether the underlying device connection is still usable. + /// + /// Implementations that can detect a permanent disconnect should override + /// this. The default preserves the behavior of transports that cannot + /// report connection state. + fn is_connected(&self) -> bool { + true + } + /// If the implementation already knows whether the underlying HID channel /// supports HID++ messages, it should return `Some((supports_short, /// supports_long))` from this method. @@ -414,6 +423,11 @@ impl HidppChannel { }) } + /// Whether the underlying HID transport still reports a live connection. + pub fn is_connected(&self) -> bool { + self.raw_channel.is_connected() + } + /// Sets the software ID that should be returned by the next call to /// [`Self::get_sw_id`]. ///