From e3a34cd2dddf1994622d74656c4d525042df8a2b Mon Sep 17 00:00:00 2001 From: Tzu-Chieh Huang Date: Fri, 17 Jul 2026 17:26:59 -0700 Subject: [PATCH] feat(alsa): add event-driven reconciliation --- crates/focusrited/src/lib.rs | 23 ++- crates/focusrited/src/main.rs | 10 +- crates/focusrited/src/scarlett2_alsa.rs | 40 ++++- crates/focusrited/src/worker.rs | 192 ++++++++++++++++++++-- crates/focusrited/tests/scarlett2_alsa.rs | 39 ++--- docs/phases/phase-3-pi-compatibility.md | 130 ++++++++++----- docs/roadmap.md | 6 +- 7 files changed, 347 insertions(+), 93 deletions(-) diff --git a/crates/focusrited/src/lib.rs b/crates/focusrited/src/lib.rs index b843b9d..c49092f 100644 --- a/crates/focusrited/src/lib.rs +++ b/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; 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 { + thread::sleep(timeout); + Ok(false) + } } #[derive(Clone, Debug, Eq, PartialEq)] @@ -168,6 +174,21 @@ impl Service { } } + /// Reconciles one hardware event immediately. A false result is a timeout. + pub fn wait_for_change(&mut self, timeout: Duration) -> Result { + 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; diff --git a/crates/focusrited/src/main.rs b/crates/focusrited/src/main.rs index 4f5df09..fc7367f 100644 --- a/crates/focusrited/src/main.rs +++ b/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(_) => {} } } } diff --git a/crates/focusrited/src/scarlett2_alsa.rs b/crates/focusrited/src/scarlett2_alsa.rs index 780d6c1..5a66fb5 100644 --- a/crates/focusrited/src/scarlett2_alsa.rs +++ b/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,6 +69,7 @@ impl std::error::Error for DiscoveryError {} pub struct Scarlett2Alsa { card: String, writable_controls: BTreeSet, + event_control: Option, } impl Scarlett2Alsa { @@ -75,6 +77,7 @@ impl Scarlett2Alsa { 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 { + 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:` and reads every ALSA control once. diff --git a/crates/focusrited/src/worker.rs b/crates/focusrited/src/worker.rs index 4874321..64d58e4 100644 --- a/crates/focusrited/src/worker.rs +++ b/crates/focusrited/src/worker.rs @@ -6,6 +6,7 @@ use std::{ mpsc::{Receiver, SyncSender, sync_channel}, }, thread::{self, JoinHandle}, + time::{Duration, Instant}, }; use crate::{ @@ -13,6 +14,9 @@ use crate::{ }; 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(mut service: Service, receiver: Receiver) { - 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(service: &mut Service, 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(service: &Service) -> State { @@ -171,7 +207,13 @@ fn state(service: &Service) -> 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, + } + + impl Device for QueuedEventDevice { + fn snapshot(&mut self) -> Result { + Ok(self.snapshot.clone()) + } + + fn write(&mut self, _: &ControlId, _: Value) -> Result<(), DeviceError> { + Ok(()) + } + + fn wait_for_change(&mut self, timeout: Duration) -> Result { + 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 { + Ok(self.snapshot.clone()) + } + + fn write(&mut self, _: &ControlId, _: Value) -> Result<(), DeviceError> { + Ok(()) + } + + fn wait_for_change(&mut self, _: Duration) -> Result { + 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::>(); + assert_eq!( + states[REQUEST_BATCH_LIMIT].snapshot.values[&control], + Value::Bool(true) + ); + assert_eq!(states[REQUEST_BATCH_LIMIT].revision, 2); + worker.stop().unwrap(); + } } diff --git a/crates/focusrited/tests/scarlett2_alsa.rs b/crates/focusrited/tests/scarlett2_alsa.rs index e293709..86d234a 100644 --- a/crates/focusrited/tests/scarlett2_alsa.rs +++ b/crates/focusrited/tests/scarlett2_alsa.rs @@ -5,9 +5,9 @@ use std::sync::Mutex; #[cfg(feature = "hardware-write-tests")] use focusrited::Device; use focusrited::{ - ControlId, DeviceError, ServiceError, Value, + ControlId, Value, scarlett2_alsa::{Scarlett2Alsa, ValueType, discover}, - worker::{DeviceWorker, WorkerError}, + worker::DeviceWorker, }; static HARDWARE_LOCK: Mutex<()> = Mutex::new(()); @@ -76,7 +76,7 @@ fn reconciles_external_direct_monitor_change() { for _ in 0..120 { std::thread::sleep(std::time::Duration::from_millis(250)); - let current = worker.refresh().unwrap(); + let current = worker.state().unwrap(); if current.snapshot.values[&control] != previous { println!( "Direct Monitor after: {:?}; revision {} to {}.", @@ -110,26 +110,19 @@ fn reconnects_after_solo_disconnect() { for _ in 0..240 { std::thread::sleep(std::time::Duration::from_millis(250)); - match worker.refresh() { - Ok(state) if saw_offline && state.online => { - println!( - "Solo reconnected; revision {} to {}.", - initial.revision, state.revision - ); - assert!(state.revision > initial.revision); - worker.stop().unwrap(); - return; - } - Ok(_) => {} - Err(WorkerError::Service(ServiceError::Device(DeviceError::Offline))) => { - if !saw_offline { - let offline = worker.state().unwrap(); - println!("Solo offline at revision {}.", offline.revision); - assert!(!offline.online); - saw_offline = true; - } - } - Err(error) => panic!("unexpected worker error: {error:?}"), + let state = worker.state().unwrap(); + if saw_offline && state.online { + println!( + "Solo reconnected; revision {} to {}.", + initial.revision, state.revision + ); + assert!(state.revision > initial.revision); + worker.stop().unwrap(); + return; + } + if !state.online && !saw_offline { + println!("Solo offline at revision {}.", state.revision); + saw_offline = true; } } diff --git a/docs/phases/phase-3-pi-compatibility.md b/docs/phases/phase-3-pi-compatibility.md index 7539bff..2efe252 100644 --- a/docs/phases/phase-3-pi-compatibility.md +++ b/docs/phases/phase-3-pi-compatibility.md @@ -2,9 +2,9 @@ ## Status -In progress. MR 1 and MR 2 complete; MR 3 is next. Target is prepared Pi OS -ARM64 with Scarlett Solo 4th Gen connected directly by USB. Phase 2 WSL -evidence remains valid only for its WSL scope. +Complete pending review and merge. Target is prepared Pi OS ARM64 with Scarlett +Solo 4th Gen connected directly by USB. Phase 2 WSL evidence remains valid +only for its WSL scope. ## Goal @@ -164,68 +164,112 @@ Read-only only. No control write or stored-profile apply. - `focusrited --card Gen` starts successfully for five seconds with an empty disposable profile-store path. It receives no command; temporary directory removal confirms no profile file was written. -- No ALSA control, profile, fixture, or adapter code changed. MR3 may capture - metadata and verify an operator-initiated external Direct Monitor change. +- No ALSA control, profile, fixture, or adapter code changed. Pi metadata + already matches the Phase 2 fixture; MR3 therefore adds event-driven + reconciliation before lifecycle recovery. -### MR 3: Native reconciliation and controlled write metadata +### MR 3: Event-driven Pi state reconciliation **Scope** -- Verify external Direct Monitor/front-panel change reaches service state and - advances revision without daemon write. -- Capture integer ranges and enum items exposed by Pi ALSA. -- Make only evidence-backed metadata/validation changes; unsupported domains - remain explicit and non-writable. +- Keep a persistent ALSA control-event source for the selected card; reconcile + external state on its events rather than using periodic full snapshots as the + normal control-sync path. +- Preserve `focusrited` as sole device owner: event handling performs no + command or ALSA write. +- Reconcile every received ALSA control event immediately. Do not throttle or + coalesce service state; each event updates authoritative state. +- GUI clients cache incoming state and cap rendering at 60 Hz. This is a UI + concern, not a service delivery limit. Meter display remains + capability-discovered until Pi ALSA evidence proves a source. +- Retain a bounded periodic health check only to detect missed events and + device loss, at a three-second interval. It is not the normal + state-synchronization mechanism. **Verification** - MR 1 checks. -- Read-only external-change test with operator interaction. -- Add or update mock tests for each parsing/validation defect found. - -**Hardware action** +- Full Rust verification: format, Clippy with warnings denied, and workspace + tests. +- Mock coverage for event ordering, high-rate event handling, missed-event + health recovery, and no-write behavior. +- Read-only Pi external-change test proves event-driven state/revision update. -Read-only daemon behavior. Operator may change front-panel Direct Monitor; -daemon must not write any control. +**Execution plan** -### MR 4: Native lifecycle recovery +1. Add the smallest persistent event loop that can service both ALSA events and + serialized worker requests. Do not add a second device owner or a polling + thread for normal state updates. +2. Reconcile event-driven changes immediately and retain a three-second health + fallback. Keep GUI cache/render throttling out of the service. +3. Add mock checks before Pi use. Then, with explicit approval for an + operator-initiated front-panel state change, prove the event path on `Gen` + without a daemon command or ALSA write. +4. Record sanitized event capability, state/revision evidence, measured update + timing, and any unsupported meter source. -**Scope** +**Hardware action** -- Verify service behavior through physical Solo unplug/replug and Pi reboot. -- Diagnose USB power, ALSA-node, service-startup, and recovery ordering issues. -- Add the smallest code or deployment change proven necessary. -- Record sanitized lifecycle evidence and known limits. +Operator front-panel Direct Monitor change requires explicit session approval. +No daemon ALSA write, routing, clock, reset, firmware, or profile apply. -**Verification** +**Pre-event evidence — 2026-07-17** -- MR 1 checks. -- Read-only physical disconnect/reconnect test. -- Read-only daemon start after reboot and recovery after reattach. +- With explicit approval, the read-only + `reconciles_external_direct_monitor_change` test observed an + operator-initiated Direct Monitor change on card `Gen`. The authoritative + service state changed and revision advanced from 1 to 2 within 30 seconds. +- The test issued no service command or daemon ALSA write, but used manual + 250 ms polling; it does not validate the planned event path. Integer ranges + and enum domains remain unchanged from the Pi MR2 fixture comparison. -**Hardware action** +**Event evidence — 2026-07-17** -Physical cable unplug/replug and Pi reboot require session approval. No ALSA -write, routing, clock, reset, firmware, or profile apply. +- With explicit approval, the same read-only test passed after its polling + refresh was removed. A front-panel Direct Monitor change on `Gen` updated + authoritative service state and advanced revision from 1 to 2. The test made + state requests only; reconciliation was initiated by the ALSA event loop. +- The test bounds detection to 30 seconds but does not measure operator-action + to event-delivery latency. Meter-event availability and rate remain Phase 4b + discovery work. +- Worker request handling is bounded to four requests per turn before a + zero-wait ALSA event check and health-check opportunity. Mock coverage proves + queued client requests cannot starve event reconciliation. -### MR 5: Phase 3 closeout +### MR 4: Pi lifecycle recovery and Phase 3 closeout **Scope** -- Consolidate Pi prerequisites and verified commands in deployment/hardware - documentation. -- Mark Phase 3 evidence and unresolved limits. -- Add only regression tests for defects actually found in MRs 1–4. +- Verify event-driven service behavior through physical Solo unplug/replug and + Pi reboot. +- Diagnose only observed USB power, ALSA-node, event-source, and startup + recovery failures. +- Add the smallest evidence-backed fix, regression check, and deployment + documentation necessary; record limits and close Phase 3 if exit checks pass. **Verification** -- Full Rust verification: format, Clippy with warnings denied, and workspace - tests. -- Repeat one bounded read-only discovery and daemon-start check on Pi. +- MR 1 checks and full Rust verification. +- Read-only physical disconnect/reconnect and post-reboot daemon-start checks. +- Repeat one bounded event-driven external-change check on Pi. **Hardware action** -Read-only only. +Physical cable unplug/replug and Pi reboot each require explicit session +approval. No ALSA write, routing, clock, reset, firmware, or profile apply. + +**Evidence — 2026-07-17 (in progress)** + +- With explicit approval, the read-only `reconnects_after_solo_disconnect` test + observed Solo offline at revision 3, then a fresh online snapshot at revision + 4 after reconnect. The test requested state only; recovery came from the + worker event/health path. +- While unplugged, two bounded ALSA card-absence diagnostics occurred during + recovery probing. The worker did not repeatedly reopen ALSA or flood logs. +- No daemon command or ALSA write occurred. +- With explicit approval, Pi rebooted successfully. Post-reboot read-only + `discovers_attached_solo` passed on card `Gen`; five-second `focusrited` + startup received no command and left its disposable profile store empty. ## Exit checks @@ -233,10 +277,10 @@ Read-only only. write tests require explicit feature plus session approval. - [x] Solo service builds and starts natively on Pi against selected ALSA card. - [x] Sanitized Pi discovery proves capabilities and documented prerequisites. -- [ ] External/front-panel change reconciles into authoritative service state. -- [ ] Physical unplug/replug recovers to a fresh online snapshot. -- [ ] Pi reboot returns daemon and Solo readiness without device-state mutation. -- [ ] Target-specific limits, commands, and unresolved issues are documented. +- [x] External/front-panel change reconciles into authoritative service state. +- [x] Physical unplug/replug recovers to a fresh online snapshot. +- [x] Pi reboot returns daemon and Solo readiness without device-state mutation. +- [x] Target-specific limits, commands, and unresolved issues are documented. ## Update rule diff --git a/docs/roadmap.md b/docs/roadmap.md index d1201be..cb72e8f 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -42,7 +42,7 @@ domains remain explicit but non-writable. Exit: mock and Solo-on-WSL tests cover supported control behavior, failure, disconnect/reconnect, and persistence; unsupported controls are explicit. -## Phase 3: Pi compatibility verification — in progress +## Phase 3: Pi compatibility verification — complete pending review and merge Execution plan: [Phase 3 Pi compatibility plan](phases/phase-3-pi-compatibility.md). @@ -52,7 +52,9 @@ target-only build, ALSA, USB, system-service, reboot, and unplug/replug issues. Cross-compilation may be introduced only if native Pi development demonstrates a real need. -MR 1 and MR 2 are complete; MR 3 is next. +MR 1 and MR 2 are complete. MR 3 adds event-driven ALSA state reconciliation; +GUI clients later cache state and cap rendering at 60 Hz. MR 4 validates +lifecycle recovery and closes Phase 3. Before adding Phase 3 hardware coverage, split Solo tests into a read-only hardware suite (discovery, external changes, reconnect) and a write-capable