feat(alsa): add event-driven reconciliation

crates/focusrited/src/lib.rs

@@ -8,7 +8,7 @@ pub mod scarlett2_alsa;
 pub mod startup;
 pub mod worker;
 
-use std::collections::BTreeMap;
+use std::{collections::BTreeMap, thread, time::Duration};
 
 #[derive(Clone, Debug, Eq, Ord, PartialEq, PartialOrd)]
 pub struct ControlId(pub String);
@@ -69,6 +69,12 @@ pub enum DeviceError {
 pub trait Device {
     fn snapshot(&mut self) -> Result<DeviceSnapshot, DeviceError>;
     fn write(&mut self, control: &ControlId, value: Value) -> Result<(), DeviceError>;
+
+    /// Waits for a hardware state-change notification without writing hardware.
+    fn wait_for_change(&mut self, timeout: Duration) -> Result<bool, DeviceError> {
+        thread::sleep(timeout);
+        Ok(false)
+    }
 }
 
 #[derive(Clone, Debug, Eq, PartialEq)]
@@ -168,6 +174,21 @@ impl<D: Device> Service<D> {
         }
     }
 
+    /// Reconciles one hardware event immediately. A false result is a timeout.
+    pub fn wait_for_change(&mut self, timeout: Duration) -> Result<bool, ServiceError> {
+        match self.device.wait_for_change(timeout) {
+            Ok(true) => {
+                self.refresh()?;
+                Ok(true)
+            }
+            Ok(false) => Ok(false),
+            Err(error) => {
+                self.mark_offline();
+                Err(ServiceError::Device(error))
+            }
+        }
+    }
+
     pub fn mark_offline(&mut self) {
         if self.online {
             self.online = false;

crates/focusrited/src/main.rs

@@ -28,20 +28,20 @@ fn main() {
 
     loop {
         thread::sleep(Duration::from_secs(1));
-        match worker.refresh() {
-            Ok(state) if !online => {
+        match worker.state() {
+            Ok(state) if state.online && !online => {
                 eprintln!(
                     "focusrited: device recovered at revision {}",
                     state.revision
                 );
                 online = true;
             }
-            Ok(_) => {}
-            Err(error) if online => {
-                eprintln!("focusrited: device offline: {error}");
+            Ok(state) if !state.online && online => {
+                eprintln!("focusrited: device offline");
                 online = false;
             }
             Err(_) => {}
+            Ok(_) => {}
         }
     }
 }

crates/focusrited/src/scarlett2_alsa.rs

@@ -3,11 +3,12 @@
 //! Instances stay read-only unless constructed with an explicitly approved
 //! discovered boolean control for a bounded hardware test.
 
-use std::collections::BTreeSet;
+use std::{collections::BTreeSet, time::Duration};
 
 use alsa::{
     ctl::{Ctl, ElemType, ElemValue},
     hctl::HCtl,
+    poll::poll_all,
 };
 
 use crate::{
@@ -68,13 +69,15 @@ impl std::error::Error for DiscoveryError {}
 pub struct Scarlett2Alsa {
     card: String,
     writable_controls: BTreeSet<ControlId>,
+    event_control: Option<Ctl>,
 }
 
 impl Scarlett2Alsa {
     pub fn new(card: impl Into<String>) -> Self {
         Self {
             card: card.into(),
             writable_controls: BTreeSet::new(),
+            event_control: None,
         }
     }
 
@@ -86,6 +89,7 @@ impl Scarlett2Alsa {
         Self {
             card: card.into(),
             writable_controls,
+            event_control: None,
         }
     }
 }
@@ -106,6 +110,40 @@ impl Device for Scarlett2Alsa {
         };
         write_boolean(&self.card, control, value)
     }
+
+    fn wait_for_change(&mut self, timeout: Duration) -> Result<bool, DeviceError> {
+        if self.event_control.is_none() {
+            let control =
+                Ctl::new(&format!("hw:{}", self.card), true).map_err(|_| DeviceError::Offline)?;
+            control
+                .subscribe_events(true)
+                .map_err(|_| DeviceError::Offline)?;
+            self.event_control = Some(control);
+        }
+
+        let timeout = timeout.as_millis().try_into().unwrap_or(i32::MAX);
+        let result = {
+            let control = self
+                .event_control
+                .as_ref()
+                .expect("event control initialized");
+            if poll_all(&[control], timeout)
+                .map_err(|_| DeviceError::Offline)?
+                .is_empty()
+            {
+                Ok(false)
+            } else {
+                control
+                    .read()
+                    .map(|event| event.is_some())
+                    .map_err(|_| DeviceError::Offline)
+            }
+        };
+        if result.is_err() {
+            self.event_control = None;
+        }
+        result
+    }
 }
 
 /// Opens `hw:<card>` and reads every ALSA control once.

crates/focusrited/src/worker.rs

@@ -6,13 +6,17 @@ use std::{
         mpsc::{Receiver, SyncSender, sync_channel},
     },
     thread::{self, JoinHandle},
+    time::{Duration, Instant},
 };
 
 use crate::{
     ControlId, Device, DeviceSnapshot, Service, ServiceError, Value, profile_store::Profiles,
 };
 
 const QUEUE_LIMIT: usize = 32;
+const EVENT_WAIT: Duration = Duration::from_millis(10);
+const HEALTH_CHECK_INTERVAL: Duration = Duration::from_secs(3);
+const REQUEST_BATCH_LIMIT: usize = 4;
 
 #[derive(Clone, Debug, Eq, PartialEq)]
 pub struct State {
@@ -138,27 +142,59 @@ impl DeviceWorker {
 }
 
 fn run<D: Device>(mut service: Service<D>, receiver: Receiver<Request>) {
-    while let Ok(request) = receiver.recv() {
-        match request {
-            Request::State(reply) => {
-                let _ = reply.send(state(&service));
-            }
-            Request::Refresh(reply) => {
-                let _ = reply.send(service.refresh().map(|_| state(&service)));
-            }
-            Request::Command {
-                control,
-                value,
-                reply,
-            } => {
-                let _ = reply.send(service.command(&control, value).map(|_| state(&service)));
-            }
-            Request::Stop(reply) => {
-                let _ = reply.send(());
+    let mut last_health_check = Instant::now();
+    loop {
+        let mut processed = 0;
+        while processed < REQUEST_BATCH_LIMIT {
+            let request = match receiver.try_recv() {
+                Ok(request) => request,
+                Err(std::sync::mpsc::TryRecvError::Empty) => break,
+                Err(std::sync::mpsc::TryRecvError::Disconnected) => return,
+            };
+            processed += 1;
+            if !handle_request(&mut service, request) {
                 return;
             }
         }
+
+        if service.is_online() {
+            let event_wait = if processed == REQUEST_BATCH_LIMIT {
+                Duration::ZERO
+            } else {
+                EVENT_WAIT
+            };
+            let _ = service.wait_for_change(event_wait);
+        } else {
+            thread::sleep(EVENT_WAIT);
+        }
+        if last_health_check.elapsed() >= HEALTH_CHECK_INTERVAL {
+            let _ = service.refresh();
+            last_health_check = Instant::now();
+        }
+    }
+}
+
+fn handle_request<D: Device>(service: &mut Service<D>, request: Request) -> bool {
+    match request {
+        Request::State(reply) => {
+            let _ = reply.send(state(service));
+        }
+        Request::Refresh(reply) => {
+            let _ = reply.send(service.refresh().map(|_| state(service)));
+        }
+        Request::Command {
+            control,
+            value,
+            reply,
+        } => {
+            let _ = reply.send(service.command(&control, value).map(|_| state(service)));
+        }
+        Request::Stop(reply) => {
+            let _ = reply.send(());
+            return false;
+        }
     }
+    true
 }
 
 fn state<D: Device>(service: &Service<D>) -> State {
@@ -171,7 +207,13 @@ fn state<D: Device>(service: &Service<D>) -> State {
 
 #[cfg(test)]
 mod tests {
-    use std::collections::BTreeMap;
+    use std::{
+        collections::BTreeMap,
+        sync::{
+            Arc,
+            atomic::{AtomicBool, Ordering},
+        },
+    };
 
     use super::*;
     use crate::{ControlCapability, DeviceError, ValueDomain};
@@ -189,6 +231,58 @@ mod tests {
         }
     }
 
+    struct EventDevice {
+        snapshot: DeviceSnapshot,
+        changed: bool,
+    }
+
+    struct QueuedEventDevice {
+        snapshot: DeviceSnapshot,
+        event_ready: Arc<AtomicBool>,
+    }
+
+    impl Device for QueuedEventDevice {
+        fn snapshot(&mut self) -> Result<DeviceSnapshot, DeviceError> {
+            Ok(self.snapshot.clone())
+        }
+
+        fn write(&mut self, _: &ControlId, _: Value) -> Result<(), DeviceError> {
+            Ok(())
+        }
+
+        fn wait_for_change(&mut self, timeout: Duration) -> Result<bool, DeviceError> {
+            if self.event_ready.swap(false, Ordering::SeqCst) {
+                self.snapshot
+                    .values
+                    .insert(ControlId("direct-monitor".into()), Value::Bool(true));
+                return Ok(true);
+            }
+            thread::sleep(timeout);
+            Ok(false)
+        }
+    }
+
+    impl Device for EventDevice {
+        fn snapshot(&mut self) -> Result<DeviceSnapshot, DeviceError> {
+            Ok(self.snapshot.clone())
+        }
+
+        fn write(&mut self, _: &ControlId, _: Value) -> Result<(), DeviceError> {
+            Ok(())
+        }
+
+        fn wait_for_change(&mut self, _: Duration) -> Result<bool, DeviceError> {
+            if self.changed {
+                return Ok(false);
+            }
+            self.changed = true;
+            self.snapshot
+                .values
+                .insert(ControlId("direct-monitor".into()), Value::Bool(true));
+            Ok(true)
+        }
+    }
+
     #[test]
     fn serial_worker_confirms_command_before_returning_state() {
         let volume = ControlId("output.volume".into());
@@ -213,4 +307,66 @@ mod tests {
         assert_eq!(state.revision, 2);
         worker.stop().unwrap();
     }
+
+    #[test]
+    fn event_reconciles_without_a_refresh_request() {
+        let control = ControlId("direct-monitor".into());
+        let worker = DeviceWorker::start(EventDevice {
+            snapshot: DeviceSnapshot {
+                device_id: "mock-device".into(),
+                capability_schema: "mock-v1".into(),
+                capabilities: Vec::new(),
+                values: BTreeMap::from([(control.clone(), Value::Bool(false))]),
+            },
+            changed: false,
+        })
+        .unwrap();
+
+        for _ in 0..10 {
+            let state = worker.state().unwrap();
+            if state.snapshot.values[&control] == Value::Bool(true) {
+                assert_eq!(state.revision, 2);
+                worker.stop().unwrap();
+                return;
+            }
+            thread::sleep(Duration::from_millis(10));
+        }
+
+        worker.stop().unwrap();
+        panic!("event did not reconcile");
+    }
+
+    #[test]
+    fn queued_requests_do_not_starve_event_reconciliation() {
+        let control = ControlId("direct-monitor".into());
+        let event_ready = Arc::new(AtomicBool::new(false));
+        let worker = DeviceWorker::start(QueuedEventDevice {
+            snapshot: DeviceSnapshot {
+                device_id: "mock-device".into(),
+                capability_schema: "mock-v1".into(),
+                capabilities: Vec::new(),
+                values: BTreeMap::from([(control.clone(), Value::Bool(false))]),
+            },
+            event_ready: event_ready.clone(),
+        })
+        .unwrap();
+        let mut replies = Vec::with_capacity(QUEUE_LIMIT);
+        for _ in 0..QUEUE_LIMIT {
+            let (sender, receiver) = std::sync::mpsc::channel();
+            worker.sender.send(Request::State(sender)).unwrap();
+            replies.push(receiver);
+        }
+        event_ready.store(true, Ordering::SeqCst);
+
+        let states = replies
+            .into_iter()
+            .map(|receiver| receiver.recv_timeout(Duration::from_secs(1)).unwrap())
+            .collect::<Vec<_>>();
+        assert_eq!(
+            states[REQUEST_BATCH_LIMIT].snapshot.values[&control],
+            Value::Bool(true)
+        );
+        assert_eq!(states[REQUEST_BATCH_LIMIT].revision, 2);
+        worker.stop().unwrap();
+    }
 }