From 23284e0babc7cc1ce3400db23163794a608e05b0 Mon Sep 17 00:00:00 2001 From: Tzu-Chieh Huang Date: Fri, 17 Jul 2026 19:52:11 -0700 Subject: [PATCH 1/3] feat(ipc): add local Unix socket API --- Cargo.lock | 98 ++++ crates/focusrited/Cargo.toml | 2 + crates/focusrited/src/ipc.rs | 651 ++++++++++++++++++++++ crates/focusrited/src/lib.rs | 14 +- crates/focusrited/src/main.rs | 9 +- crates/focusrited/src/scarlett2_alsa.rs | 2 +- crates/focusrited/src/startup.rs | 18 +- docs/phases/phase-3-pi-compatibility.md | 11 +- docs/phases/phase-4a-local-touchscreen.md | 154 +++++ docs/protocol.md | 21 + docs/roadmap.md | 11 +- 11 files changed, 972 insertions(+), 19 deletions(-) create mode 100644 crates/focusrited/src/ipc.rs create mode 100644 docs/phases/phase-4a-local-touchscreen.md diff --git a/Cargo.lock b/Cargo.lock index 12fd136..168073a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -58,22 +58,120 @@ name = "focusrited" version = "0.1.0" dependencies = [ "alsa", + "serde", + "serde_json", ] +[[package]] +name = "itoa" +version = "1.0.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682" + [[package]] name = "libc" version = "0.2.186" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "68ab91017fe16c622486840e4c83c9a37afeff978bd239b5293d61ece587de66" +[[package]] +name = "memchr" +version = "2.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98" + [[package]] name = "pkg-config" version = "0.3.33" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "19f132c84eca552bf34cab8ec81f1c1dcc229b811638f9d283dceabe58c5569e" +[[package]] +name = "proc-macro2" +version = "1.0.106" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8fd00f0bb2e90d81d1044c2b32617f68fcb9fa3bb7640c23e9c748e53fb30934" +dependencies = [ + "unicode-ident", +] + +[[package]] +name = "quote" +version = "1.0.46" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dfbc457d0c7a0759a614551b11a6409e5951f6c7537be1f1b7682b9ae9230368" +dependencies = [ + "proc-macro2", +] + +[[package]] +name = "serde" +version = "1.0.228" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9a8e94ea7f378bd32cbbd37198a4a91436180c5bb472411e48b5ec2e2124ae9e" +dependencies = [ + "serde_core", + "serde_derive", +] + +[[package]] +name = "serde_core" +version = "1.0.228" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "41d385c7d4ca58e59fc732af25c3983b67ac852c1a25000afe1175de458b67ad" +dependencies = [ + "serde_derive", +] + +[[package]] +name = "serde_derive" +version = "1.0.228" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d540f220d3187173da220f885ab66608367b6574e925011a9353e4badda91d79" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "serde_json" +version = "1.0.150" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e8014e44b4736ed0538adeecded0fce2a272f22dc9578a7eb6b2d9993c74cfb9" +dependencies = [ + "itoa", + "memchr", + "serde", + "serde_core", + "zmij", +] + [[package]] name = "shlex" version = "2.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f8fadd59c855ef2080decdef8ff161eb6661b86933c9d82e5ba29dc602a55aba" + +[[package]] +name = "syn" +version = "2.0.119" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "872831b642d1a07999a962a351ed35b955ea2cfc8f3862091e2a240a84f17297" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + +[[package]] +name = "unicode-ident" +version = "1.0.24" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75" + +[[package]] +name = "zmij" +version = "1.0.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "29666d0abbfad1e3dc4dcf6144730dd3a3ab225bbbdac83319345b1b44ccfc1b" diff --git a/crates/focusrited/Cargo.toml b/crates/focusrited/Cargo.toml index 77138a7..abc0b48 100644 --- a/crates/focusrited/Cargo.toml +++ b/crates/focusrited/Cargo.toml @@ -10,3 +10,5 @@ hardware-write-tests = [] [dependencies] alsa = "0.12.0" +serde = { version = "1.0.228", features = ["derive"] } +serde_json = "1.0.150" diff --git a/crates/focusrited/src/ipc.rs b/crates/focusrited/src/ipc.rs new file mode 100644 index 0000000..86e9f0d --- /dev/null +++ b/crates/focusrited/src/ipc.rs @@ -0,0 +1,651 @@ +//! Bounded local Unix-socket API for local clients. + +use std::{ + collections::VecDeque, + fs, + io::{self, Read, Write}, + os::unix::{ + fs::{FileTypeExt, PermissionsExt}, + net::{UnixListener, UnixStream}, + }, + path::{Path, PathBuf}, + sync::{ + Arc, + atomic::{AtomicBool, Ordering}, + }, + thread::{self, JoinHandle}, + time::{Duration, SystemTime, UNIX_EPOCH}, +}; + +use serde::{Deserialize, Serialize}; + +use crate::{ControlId, DeviceSnapshot, ServiceError, Value, worker::DeviceWorker}; + +const PROTOCOL_VERSION: u8 = 1; +const MAX_MESSAGE_BYTES: usize = 64 * 1024; +const MAX_OUTBOUND_BYTES: usize = 64 * 1024; +const REQUEST_BATCH_LIMIT: usize = 4; +const LOOP_WAIT: Duration = Duration::from_millis(20); + +pub struct LocalServer { + running: Arc, + thread: Option>, + path: PathBuf, +} + +impl LocalServer { + pub fn start(worker: Arc, path: PathBuf) -> io::Result { + if let Some(parent) = path.parent() { + fs::create_dir_all(parent)?; + } + remove_stale_socket(&path)?; + let listener = UnixListener::bind(&path)?; + fs::set_permissions(&path, fs::Permissions::from_mode(0o660))?; + listener.set_nonblocking(true)?; + let running = Arc::new(AtomicBool::new(true)); + let thread_running = Arc::clone(&running); + let thread_path = path.clone(); + let thread = thread::spawn(move || { + run(listener, worker, thread_running); + let _ = fs::remove_file(thread_path); + }); + Ok(Self { + running, + thread: Some(thread), + path, + }) + } + + pub fn stop(mut self) { + self.running.store(false, Ordering::SeqCst); + if let Some(thread) = self.thread.take() { + let _ = thread.join(); + } + } +} + +impl Drop for LocalServer { + fn drop(&mut self) { + self.running.store(false, Ordering::SeqCst); + if let Some(thread) = self.thread.take() { + let _ = thread.join(); + } + let _ = fs::remove_file(&self.path); + } +} + +fn remove_stale_socket(path: &Path) -> io::Result<()> { + match fs::symlink_metadata(path) { + Ok(metadata) if metadata.file_type().is_socket() => fs::remove_file(path), + Ok(_) => Err(io::Error::new( + io::ErrorKind::AlreadyExists, + "socket path exists and is not a socket", + )), + Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(()), + Err(error) => Err(error), + } +} + +fn run(listener: UnixListener, worker: Arc, running: Arc) { + let instance_id = format!( + "{}-{}", + std::process::id(), + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_nanos() + ); + let mut clients = Vec::new(); + let mut last_revision = None; + while running.load(Ordering::SeqCst) { + accept_clients(&listener, &worker, &instance_id, &mut clients); + if let Ok(state) = worker.state() + && last_revision + .replace(state.revision) + .is_some_and(|revision| revision != state.revision) + { + broadcast( + &mut clients, + Outbound::Event(state_message(&instance_id, "event", state)), + ); + } + clients.retain_mut(|client| { + let readable = read_client(client, &worker, &instance_id); + let written = flush_client(client); + readable && written && !client.closing + }); + thread::sleep(LOOP_WAIT); + } +} + +fn accept_clients( + listener: &UnixListener, + worker: &Arc, + instance_id: &str, + clients: &mut Vec, +) { + loop { + let stream = match listener.accept() { + Ok((stream, _)) => stream, + Err(error) if error.kind() == io::ErrorKind::WouldBlock => return, + Err(_) => return, + }; + if stream.set_nonblocking(true).is_err() { + continue; + } + let mut client = Client::new(stream); + let message = match worker.state() { + Ok(state) => state_message(instance_id, "snapshot", state), + Err(_) => error_message("worker_stopped"), + }; + if client.queue(Outbound::Reply(message)) { + clients.push(client); + } + } +} + +fn read_client(client: &mut Client, worker: &Arc, instance_id: &str) -> bool { + let mut bytes = [0; 4096]; + match client.stream.read(&mut bytes) { + Ok(0) => return false, + Ok(count) => { + client.input.extend_from_slice(&bytes[..count]); + if client.input.len() > MAX_MESSAGE_BYTES { + return client.reject("message_too_large"); + } + } + Err(error) if error.kind() == io::ErrorKind::WouldBlock => {} + Err(_) => return false, + } + for _ in 0..REQUEST_BATCH_LIMIT { + let Some(end) = client.input.iter().position(|byte| *byte == b'\n') else { + break; + }; + let line = client.input.drain(..=end).collect::>(); + let line = &line[..line.len() - 1]; + match serde_json::from_slice::(line) { + Ok(request) if request.version == PROTOCOL_VERSION => { + if !handle_request(client, worker, instance_id, request) { + return false; + } + } + Ok(_) => { + return client.reject("unsupported_version"); + } + Err(_) => { + return client.reject("malformed_message"); + } + } + } + true +} + +fn handle_request( + client: &mut Client, + worker: &Arc, + instance_id: &str, + request: Request, +) -> bool { + let message = match request.kind.as_str() { + "snapshot" => match worker.state() { + Ok(state) => state_message(instance_id, "snapshot", state), + Err(_) => error_message("worker_stopped"), + }, + "command" => match (request.control, request.value) { + (Some(control), Some(value)) => match worker.command(ControlId(control), value) { + Ok(state) => state_message(instance_id, "command_result", state), + Err(error) => error_message(service_error(error)), + }, + _ => error_message("invalid_command"), + }, + _ => error_message("unknown_message_type"), + }; + client.queue(Outbound::Reply(message)) +} + +fn flush_client(client: &mut Client) -> bool { + while let Some(message) = client.outbound.front_mut() { + match client.stream.write(&message.bytes[message.offset..]) { + Ok(0) => return false, + Ok(count) => message.offset += count, + Err(error) if error.kind() == io::ErrorKind::WouldBlock => return true, + Err(_) => return false, + } + if let Some(message) = client + .outbound + .pop_front_if(|message| message.offset == message.bytes.len()) + { + client.outbound_bytes -= message.bytes.len(); + } + } + true +} + +fn broadcast(clients: &mut [Client], message: Outbound) { + for client in clients { + let _ = client.queue(message.clone()); + } +} + +struct Client { + stream: UnixStream, + input: Vec, + outbound: VecDeque, + outbound_bytes: usize, + closing: bool, +} + +impl Client { + fn new(stream: UnixStream) -> Self { + Self { + stream, + input: Vec::new(), + outbound: VecDeque::new(), + outbound_bytes: 0, + closing: false, + } + } + + fn queue(&mut self, outbound: Outbound) -> bool { + let bytes = match serde_json::to_vec(outbound.message()) { + Ok(mut bytes) => { + bytes.push(b'\n'); + bytes + } + Err(_) => return false, + }; + if bytes.len() > MAX_MESSAGE_BYTES { + return false; + } + if matches!(outbound, Outbound::Event(_)) + && let Some(message) = self + .outbound + .iter_mut() + .find(|message| message.event && message.offset == 0) + { + self.outbound_bytes -= message.bytes.len(); + message.bytes = bytes; + message.offset = 0; + self.outbound_bytes += message.bytes.len(); + return self.outbound_bytes <= MAX_OUTBOUND_BYTES; + } + if self.outbound_bytes + bytes.len() > MAX_OUTBOUND_BYTES { + return false; + } + self.outbound_bytes += bytes.len(); + self.outbound.push_back(Message { + bytes, + offset: 0, + event: matches!(outbound, Outbound::Event(_)), + }); + true + } + + fn reject(&mut self, error: &'static str) -> bool { + self.closing = true; + self.queue(Outbound::Reply(error_message(error))) + } +} + +struct Message { + bytes: Vec, + offset: usize, + event: bool, +} + +#[derive(Clone)] +enum Outbound { + Event(Response), + Reply(Response), +} + +impl Outbound { + fn message(&self) -> &Response { + match self { + Self::Event(message) | Self::Reply(message) => message, + } + } +} + +#[derive(Deserialize)] +struct Request { + #[serde(rename = "v")] + version: u8, + #[serde(rename = "type")] + kind: String, + control: Option, + value: Option, +} + +#[derive(Clone, Serialize)] +struct Response { + v: u8, + #[serde(rename = "type")] + kind: &'static str, + #[serde(skip_serializing_if = "Option::is_none")] + instance_id: Option, + #[serde(skip_serializing_if = "Option::is_none")] + revision: Option, + #[serde(skip_serializing_if = "Option::is_none")] + online: Option, + #[serde(skip_serializing_if = "Option::is_none")] + snapshot: Option, + #[serde(skip_serializing_if = "Option::is_none")] + error: Option<&'static str>, +} + +fn state_message(instance_id: &str, kind: &'static str, state: crate::worker::State) -> Response { + Response { + v: PROTOCOL_VERSION, + kind, + instance_id: Some(instance_id.into()), + revision: Some(state.revision), + online: Some(state.online), + snapshot: Some(state.snapshot), + error: None, + } +} + +fn error_message(error: &'static str) -> Response { + Response { + v: PROTOCOL_VERSION, + kind: "error", + instance_id: None, + revision: None, + online: None, + snapshot: None, + error: Some(error), + } +} + +fn service_error(error: crate::worker::WorkerError) -> &'static str { + match error { + crate::worker::WorkerError::Stopped => "worker_stopped", + crate::worker::WorkerError::Service(ServiceError::Device(_)) => "device_error", + crate::worker::WorkerError::Service(ServiceError::UnknownControl) => "unknown_control", + crate::worker::WorkerError::Service(ServiceError::Unavailable) => "unavailable", + crate::worker::WorkerError::Service(ServiceError::ReadOnly) => "read_only", + crate::worker::WorkerError::Service(ServiceError::InvalidValue) => "invalid_value", + crate::worker::WorkerError::Service(ServiceError::UnconfirmedWrite) => "unconfirmed_write", + crate::worker::WorkerError::Service(_) => "unsupported_command", + } +} + +#[cfg(test)] +mod tests { + use std::{ + collections::BTreeMap, + io::{BufRead, BufReader}, + sync::atomic::AtomicBool, + time::Duration, + }; + + use super::*; + use crate::{ControlCapability, Device, DeviceError, ValueDomain}; + + struct MockDevice(DeviceSnapshot); + + impl Device for MockDevice { + fn snapshot(&mut self) -> Result { + Ok(self.0.clone()) + } + + fn write(&mut self, control: &ControlId, value: Value) -> Result<(), DeviceError> { + self.0.values.insert(control.clone(), value); + Ok(()) + } + } + + struct EventDevice { + snapshot: DeviceSnapshot, + event_ready: Arc, + } + + 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, timeout: Duration) -> Result { + if self.event_ready.swap(false, Ordering::SeqCst) { + self.snapshot + .values + .insert(ControlId("output.level".into()), Value::Integer(75)); + return Ok(true); + } + thread::sleep(timeout); + Ok(false) + } + } + + fn server() -> (LocalServer, Arc, PathBuf) { + let control = ControlId("output.level".into()); + let worker = Arc::new( + DeviceWorker::start(MockDevice(DeviceSnapshot { + device_id: "mock-device".into(), + capability_schema: "mock-v1".into(), + capabilities: vec![ControlCapability { + id: control.clone(), + domain: ValueDomain::Integer, + writable: true, + available: true, + minimum: Some(0), + maximum: Some(100), + }], + values: BTreeMap::from([(control, Value::Integer(50))]), + })) + .unwrap(), + ); + let path = std::env::temp_dir().join(format!( + "focusrited-ipc-test-{}-{}.sock", + std::process::id(), + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_nanos() + )); + let server = LocalServer::start(Arc::clone(&worker), path.clone()).unwrap(); + (server, worker, path) + } + + fn event_server() -> (LocalServer, Arc, PathBuf, Arc) { + let event_ready = Arc::new(AtomicBool::new(false)); + let control = ControlId("output.level".into()); + let worker = Arc::new( + DeviceWorker::start(EventDevice { + snapshot: DeviceSnapshot { + device_id: "mock-device".into(), + capability_schema: "mock-v1".into(), + capabilities: vec![ControlCapability { + id: control.clone(), + domain: ValueDomain::Integer, + writable: false, + available: true, + minimum: Some(0), + maximum: Some(100), + }], + values: BTreeMap::from([(control, Value::Integer(50))]), + }, + event_ready: Arc::clone(&event_ready), + }) + .unwrap(), + ); + let path = std::env::temp_dir().join(format!( + "focusrited-ipc-event-test-{}-{}.sock", + std::process::id(), + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_nanos() + )); + let server = LocalServer::start(Arc::clone(&worker), path.clone()).unwrap(); + (server, worker, path, event_ready) + } + + fn read_json(reader: &mut BufReader) -> serde_json::Value { + let mut line = String::new(); + reader.read_line(&mut line).unwrap(); + serde_json::from_str(&line).unwrap() + } + + fn read_type(reader: &mut BufReader, expected: &str) -> serde_json::Value { + for _ in 0..4 { + let message = read_json(reader); + if message["type"] == expected { + return message; + } + } + panic!("did not receive {expected}"); + } + + #[test] + fn snapshots_then_confirms_commands_over_local_socket() { + let (server, worker, path) = server(); + let mut stream = UnixStream::connect(&path).unwrap(); + stream + .set_read_timeout(Some(Duration::from_secs(1))) + .unwrap(); + let mut reader = BufReader::new(stream.try_clone().unwrap()); + + let snapshot = read_json(&mut reader); + assert_eq!(snapshot["type"], "snapshot"); + assert_eq!(snapshot["revision"], 1); + stream + .write_all( + b"{\"v\":1,\"type\":\"command\",\"control\":\"output.level\",\"value\":{\"type\":\"integer\",\"value\":75}}\n", + ) + .unwrap(); + let result = read_type(&mut reader, "command_result"); + assert_eq!(result["revision"], 2); + assert_eq!(result["snapshot"]["values"]["output.level"]["value"], 75); + + server.stop(); + worker.stop().unwrap(); + } + + #[test] + fn two_clients_converge_on_confirmed_state() { + let (server, worker, path) = server(); + let mut first = UnixStream::connect(&path).unwrap(); + let mut second = UnixStream::connect(&path).unwrap(); + for stream in [&first, &second] { + stream + .set_read_timeout(Some(Duration::from_secs(1))) + .unwrap(); + } + let mut first_reader = BufReader::new(first.try_clone().unwrap()); + let mut second_reader = BufReader::new(second.try_clone().unwrap()); + let _ = read_type(&mut first_reader, "snapshot"); + let _ = read_type(&mut second_reader, "snapshot"); + + first + .write_all( + b"{\"v\":1,\"type\":\"command\",\"control\":\"output.level\",\"value\":{\"type\":\"integer\",\"value\":60}}\n", + ) + .unwrap(); + assert_eq!( + read_type(&mut first_reader, "command_result")["revision"], + 2 + ); + assert_eq!(read_type(&mut first_reader, "event")["revision"], 2); + assert_eq!(read_type(&mut second_reader, "event")["revision"], 2); + second + .write_all( + b"{\"v\":1,\"type\":\"command\",\"control\":\"output.level\",\"value\":{\"type\":\"integer\",\"value\":80}}\n", + ) + .unwrap(); + assert_eq!( + read_type(&mut second_reader, "command_result")["revision"], + 3 + ); + let event = read_type(&mut first_reader, "event"); + assert_eq!(event["revision"], 3); + assert_eq!(event["snapshot"]["values"]["output.level"]["value"], 80); + + server.stop(); + worker.stop().unwrap(); + } + + #[test] + fn external_worker_event_reaches_local_client() { + let (server, worker, path, event_ready) = event_server(); + let stream = UnixStream::connect(&path).unwrap(); + stream + .set_read_timeout(Some(Duration::from_secs(1))) + .unwrap(); + let mut reader = BufReader::new(stream); + let _ = read_type(&mut reader, "snapshot"); + event_ready.store(true, Ordering::SeqCst); + + let event = read_type(&mut reader, "event"); + assert_eq!(event["revision"], 2); + assert_eq!(event["snapshot"]["values"]["output.level"]["value"], 75); + + server.stop(); + worker.stop().unwrap(); + } + + #[test] + fn unsent_events_coalesce_and_queue_overflow_rejects_client() { + let (stream, _) = UnixStream::pair().unwrap(); + let mut client = Client::new(stream); + let event = |revision| { + Outbound::Event(Response { + v: PROTOCOL_VERSION, + kind: "event", + instance_id: Some("test".into()), + revision: Some(revision), + online: Some(true), + snapshot: None, + error: None, + }) + }; + assert!(client.queue(event(1))); + assert!(client.queue(event(2))); + assert_eq!(client.outbound.len(), 1); + assert!( + client.outbound[0] + .bytes + .windows(12) + .any(|bytes| bytes == b"\"revision\":2") + ); + + let large_reply = || { + Outbound::Reply(Response { + v: PROTOCOL_VERSION, + kind: "snapshot", + instance_id: Some("x".repeat(8 * 1024)), + revision: None, + online: None, + snapshot: None, + error: None, + }) + }; + while client.queue(large_reply()) {} + assert!(client.outbound_bytes <= MAX_OUTBOUND_BYTES); + } + + #[test] + fn malformed_message_gets_error_then_disconnect() { + let (server, worker, path) = server(); + let mut stream = UnixStream::connect(&path).unwrap(); + stream + .set_read_timeout(Some(Duration::from_secs(1))) + .unwrap(); + let mut reader = BufReader::new(stream.try_clone().unwrap()); + let _ = read_json(&mut reader); + stream.write_all(b"not json\n").unwrap(); + let error = read_json(&mut reader); + assert_eq!(error["type"], "error"); + assert_eq!(error["error"], "malformed_message"); + let mut line = String::new(); + assert_eq!(reader.read_line(&mut line).unwrap(), 0); + + server.stop(); + worker.stop().unwrap(); + } +} diff --git a/crates/focusrited/src/lib.rs b/crates/focusrited/src/lib.rs index c49092f..4d02441 100644 --- a/crates/focusrited/src/lib.rs +++ b/crates/focusrited/src/lib.rs @@ -3,6 +3,7 @@ //! Linux/ALSA adapters and client transports sit outside this module. Keeping //! the policy here makes discovery and state rules testable without hardware. +pub mod ipc; pub mod profile_store; pub mod scarlett2_alsa; pub mod startup; @@ -10,10 +11,12 @@ pub mod worker; use std::{collections::BTreeMap, thread, time::Duration}; -#[derive(Clone, Debug, Eq, Ord, PartialEq, PartialOrd)] +#[derive(Clone, Debug, Eq, Ord, PartialEq, PartialOrd, serde::Deserialize, serde::Serialize)] +#[serde(transparent)] pub struct ControlId(pub String); -#[derive(Clone, Debug, Eq, PartialEq)] +#[derive(Clone, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize)] +#[serde(tag = "type", content = "value", rename_all = "snake_case")] pub enum Value { Bool(bool), Integer(i32), @@ -21,7 +24,8 @@ pub enum Value { Array(Vec), } -#[derive(Clone, Debug, Eq, PartialEq)] +#[derive(Clone, Debug, Eq, PartialEq, serde::Serialize)] +#[serde(rename_all = "snake_case")] pub enum ValueDomain { Boolean, Integer, @@ -30,7 +34,7 @@ pub enum ValueDomain { Array, } -#[derive(Clone, Debug, Eq, PartialEq)] +#[derive(Clone, Debug, Eq, PartialEq, serde::Serialize)] pub struct ControlCapability { pub id: ControlId, pub domain: ValueDomain, @@ -40,7 +44,7 @@ pub struct ControlCapability { pub maximum: Option, } -#[derive(Clone, Debug, Eq, PartialEq)] +#[derive(Clone, Debug, Eq, PartialEq, serde::Serialize)] pub struct DeviceSnapshot { /// Opaque adapter-provided identity. Profiles never cross this boundary. pub device_id: String, diff --git a/crates/focusrited/src/main.rs b/crates/focusrited/src/main.rs index fc7367f..b8d896b 100644 --- a/crates/focusrited/src/main.rs +++ b/crates/focusrited/src/main.rs @@ -1,8 +1,9 @@ //! Read-only Focusrite controller daemon bootstrap. -use std::{env, process, thread, time::Duration}; +use std::{env, process, sync::Arc, thread, time::Duration}; use focusrited::{ + ipc::LocalServer, scarlett2_alsa::Scarlett2Alsa, startup::{Config, ConfigError, load_profiles}, worker::DeviceWorker, @@ -22,7 +23,11 @@ fn main() { .as_deref() .unwrap_or_else(|| fail("--card is required")); let profiles = load_profiles(&config).unwrap_or_else(|error| fail(error)); - let worker = DeviceWorker::start_with_profiles(Scarlett2Alsa::new(card), profiles) + let worker = Arc::new( + DeviceWorker::start_with_profiles(Scarlett2Alsa::new(card), profiles) + .unwrap_or_else(|error| fail(error)), + ); + let _ipc = LocalServer::start(Arc::clone(&worker), config.socket_path) .unwrap_or_else(|error| fail(error)); let mut online = true; diff --git a/crates/focusrited/src/scarlett2_alsa.rs b/crates/focusrited/src/scarlett2_alsa.rs index 5a66fb5..dd3f4ed 100644 --- a/crates/focusrited/src/scarlett2_alsa.rs +++ b/crates/focusrited/src/scarlett2_alsa.rs @@ -290,7 +290,7 @@ fn write_boolean(card: &str, control: &ControlId, value: bool) -> Result<(), Dev let numid = control .0 .strip_prefix("alsa-numid:") - .and_then(|number| number.parse().ok()) + .and_then(|number| number.parse::().ok()) .ok_or(DeviceError::WriteDisabled)?; let hctl = HCtl::new(&format!("hw:{card}"), false).map_err(|_| DeviceError::Offline)?; hctl.load().map_err(|_| DeviceError::Offline)?; diff --git a/crates/focusrited/src/startup.rs b/crates/focusrited/src/startup.rs index a8d87d5..a41ae0d 100644 --- a/crates/focusrited/src/startup.rs +++ b/crates/focusrited/src/startup.rs @@ -8,11 +8,13 @@ use crate::{ }; pub const DEFAULT_PROFILE_STORE_PATH: &str = "/var/lib/focusrited/profiles"; +pub const DEFAULT_SOCKET_PATH: &str = "/run/focusrited/focusrited.sock"; #[derive(Clone, Debug, Eq, PartialEq)] pub struct Config { pub card: Option, pub profile_store_path: PathBuf, + pub socket_path: PathBuf, } impl Default for Config { @@ -20,6 +22,7 @@ impl Default for Config { Self { card: None, profile_store_path: DEFAULT_PROFILE_STORE_PATH.into(), + socket_path: DEFAULT_SOCKET_PATH.into(), } } } @@ -40,6 +43,12 @@ impl Config { .map(PathBuf::from) .ok_or(ConfigError::MissingProfileStorePath)?; } + "--socket" => { + config.socket_path = arguments + .next() + .map(PathBuf::from) + .ok_or(ConfigError::MissingSocketPath)?; + } "--help" | "-h" => return Err(ConfigError::Help), _ => return Err(ConfigError::UnknownArgument(argument)), } @@ -53,17 +62,18 @@ pub enum ConfigError { Help, MissingCard, MissingProfileStorePath, + MissingSocketPath, UnknownArgument(String), } impl std::fmt::Display for ConfigError { fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match self { - Self::Help => { - formatter.write_str("usage: focusrited --card CARD [--profile-store PATH]") - } + Self::Help => formatter + .write_str("usage: focusrited --card CARD [--profile-store PATH] [--socket PATH]"), Self::MissingCard => formatter.write_str("--card requires an ALSA card name"), Self::MissingProfileStorePath => formatter.write_str("--profile-store requires a path"), + Self::MissingSocketPath => formatter.write_str("--socket requires a path"), Self::UnknownArgument(argument) => write!(formatter, "unknown argument: {argument}"), } } @@ -121,6 +131,7 @@ mod tests { .unwrap(); assert_eq!(config.card.as_deref(), Some("Solo")); assert_eq!(config.profile_store_path, PathBuf::from("/tmp/profiles")); + assert_eq!(config.socket_path, PathBuf::from(DEFAULT_SOCKET_PATH)); } struct MockDevice(DeviceSnapshot); @@ -160,6 +171,7 @@ mod tests { let config = Config { card: None, profile_store_path: path.clone(), + socket_path: DEFAULT_SOCKET_PATH.into(), }; let store = ProfileStore::new(&path); let mut saved = Service::connect(device(50)).unwrap(); diff --git a/docs/phases/phase-3-pi-compatibility.md b/docs/phases/phase-3-pi-compatibility.md index 2efe252..0dbf44f 100644 --- a/docs/phases/phase-3-pi-compatibility.md +++ b/docs/phases/phase-3-pi-compatibility.md @@ -2,9 +2,9 @@ ## Status -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. +Complete. MRs 1–4 are merged and confirmed. 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 @@ -258,7 +258,7 @@ No daemon ALSA write, routing, clock, reset, firmware, or profile apply. 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)** +**Evidence — 2026-07-17 (complete)** - With explicit approval, the read-only `reconnects_after_solo_disconnect` test observed Solo offline at revision 3, then a fresh online snapshot at revision @@ -270,6 +270,9 @@ approval. No ALSA write, routing, clock, reset, firmware, or profile apply. - 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. +- MRs 1–4 were merged and confirmed after their recorded checks and hardware + sessions. Phase 3 is closed; no device-state mutation was required for its + final lifecycle evidence. ## Exit checks diff --git a/docs/phases/phase-4a-local-touchscreen.md b/docs/phases/phase-4a-local-touchscreen.md new file mode 100644 index 0000000..93fc5bf --- /dev/null +++ b/docs/phases/phase-4a-local-touchscreen.md @@ -0,0 +1,154 @@ +# Phase 4a: Local Touchscreen + +## Status + +MR 1 is ready for review and merge on `phase-4a-ipc`. Mock verification and +Pi read-only socket smoke validation are complete. Phase 3 is complete. Do not +begin UI implementation until local API review is accepted. + +## Goal + +Run a fullscreen touchscreen client on Pi using only `focusrited`'s local +Unix-socket API. The daemon remains sole policy and hardware authority. A +client crash, restart, or slow client must not interrupt daemon or device +operation. + +## Scope and guardrails + +- Local Unix-domain socket only. No TCP listener, browser, LAN authentication, + USB, or ALSA access in the client. +- Use versioned, newline-delimited JSON messages. One message is at most 64 KiB; + malformed, unsupported-version, or oversized input receives a bounded error + then its connection closes. +- Socket path and permissions are daemon-owned. Initial default is + `/run/focusrited/focusrited.sock` with owner/group mode `0660`; Phase 8 + packaging creates dedicated group and runtime directory. +- Keep one serial `DeviceWorker`. IPC handlers submit worker requests; they + never own hardware or bypass validation. +- Each connection has bounded outbound queue. State events may coalesce to its + newest revision for a slow client; command replies never coalesce. Queue + overflow disconnects only that client. +- Snapshot is resync authority. On reconnect, revision gap, or changed daemon + `instance_id`, client discards cache and requests a new snapshot. +- No profile apply or hardware write during test setup without explicit + approval. Mock tests cover mutating commands; Pi hardware validation starts + read-only. +- Direct dependencies `serde` and `serde_json` are reviewed as + MIT OR Apache-2.0 and compatible with project MIT distribution. + +## Merge-request plan + +### MR 1: Versioned local IPC transport + +**Scope** + +- Add transport-neutral `v1` wire types for snapshot request/reply, control + command/reply, state event, and bounded error. JSON field names are stable; + wire values are explicit rather than Rust debug output. +- Add daemon-generated `instance_id`, per-client connection loop, Unix listener, + signal-safe socket cleanup, and startup option for socket path. +- Send a full snapshot after successful connect and on explicit snapshot + request. Broadcast authoritative state after worker-observed revision change, + including external events and offline/recovery transitions. +- Serialize commands through existing `DeviceWorker`; return confirmed state or + a mapped safe error. Do not add profiles, idempotency keys, dangerous-control + confirmation, or compound commands yet: current service cannot implement + them correctly. +- Add mock IPC integration tests for framing, malformed input, command order, + reconnect/resync, external-state event delivery, slow-client isolation, and + two local clients converging after sequential writes. + +**Verification** + +- `cargo fmt --check` +- `cargo clippy --workspace --all-targets -- -D warnings` +- `cargo test --workspace` +- Mock integration tests exercise Unix sockets without ALSA hardware. +- On Pi, start daemon with a disposable socket/runtime directory and perform + snapshot-only connect/disconnect check. No command or profile apply. + +**Hardware action** + +Read-only Pi socket smoke test only. Any touchscreen display work or hardware +write needs separate explicit approval. + +**Evidence — 2026-07-17 (in progress)** + +- With explicit approval, `focusrited` started on Pi against ALSA ID `Gen` + using disposable socket and profile-store paths. A local `v1` snapshot + request received the automatic connection snapshot and explicit snapshot; + both were online at revision 1 with the same instance ID. +- The daemon received no command, performed no ALSA write, and its disposable + profile-store path remained absent. It then stopped cleanly. +- The current ALSA card index was 1 after reboot; named ID `Gen` remained the + stable selector. No raw control values or device identifiers are retained. +- Mock IPC coverage passes for framing errors, command ordering, reconnect + snapshots, external worker events, two-client convergence, event coalescing, + slow-client queue overflow, and bounded per-client request turns. + +### MR 2: Fullscreen touch client and primary controls + +**Scope** + +- Add smallest Rust touchscreen executable using MR 1 API only. Select UI + toolkit after confirming it builds and runs fullscreen on target Pi and fits + actual screen resolution/orientation; do not add a web stack. +- Render connection/device status plus capability-discovered writable primary + monitor/output controls. Prefer daemon-provided metadata; never hardcode + Solo control IDs or channel count. +- Use large touch targets, show confirmed values, rate-limit rendering to + 60 Hz, and resync from snapshot after reconnect or event gap. Controls absent + from capabilities remain absent, not disabled guesses. + +**Verification** + +- MR 1 checks plus UI unit/mock API tests. +- Pi fullscreen launch, touch hit-target/screen-fit check, client kill/restart, + and daemon continuity check. +- Any command test uses mock device first; live writes require explicit approval. + +**Hardware action** + +Display and touch interaction. Live control write only with explicit approval. + +### MR 3: Local profile workflow + +**Scope** + +- Complete service-side profile operations before exposing them: named save and + list, device/schema binding result, deterministic dry-run diff, explicit + reviewed apply, and per-control applied/skipped/failed report. +- Persist profile changes through existing store atomically. No auto-apply, + rollback, or profile write on daemon startup/reconnect. +- Extend local IPC and touchscreen only after daemon behavior and mock tests + prove those semantics. + +**Verification** + +- MR 1 checks plus mock coverage for binding mismatch, unavailable controls, + ordered partial failure, reconnect, and concurrent local clients. +- Pi profile exercise only after explicit approval; preserve and restore device + state through a reviewed test plan. + +**Hardware action** + +Profile save may write only service storage. Profile apply writes hardware and +requires explicit approval. + +## Exit checks + +- [ ] Daemon exposes versioned, bounded local snapshot/command/event API. +- [ ] Socket clients cannot access USB or ALSA and cannot interrupt daemon on + failure. +- [ ] Mock IPC tests cover ordering, reconnect/resync, and concurrent local + client updates. +- [ ] Fullscreen Pi touchscreen exposes only capability-discovered controls and + remains usable after client restart. +- [ ] Local profile save/list/dry-run/reviewed apply returns safe per-control + result and never auto-applies. + +## Update rule + +After each MR, record completed verification, screen/runtime findings, explicit +hardware approvals, and sanitized evidence. Do not expand into metering, LAN, +packaging, or FCP device support. diff --git a/docs/protocol.md b/docs/protocol.md index d8afe23..a2b83cc 100644 --- a/docs/protocol.md +++ b/docs/protocol.md @@ -17,6 +17,27 @@ not Rust internals. Unix-socket clients use equivalent snapshot, command, and event messages. +## Local IPC v1 + +Phase 4a local IPC is newline-delimited JSON over the daemon-owned Unix socket. +Each message is at most 64 KiB. Client requests include `v: 1` and a `type`: +`snapshot`, or `command` with opaque `control` and typed `value`. The daemon +returns `snapshot`, `command_result`, `event`, or bounded `error` JSON. State +messages carry `instance_id`, `revision`, `online`, and full authoritative +snapshot. Typed values use tagged JSON, for example +`{"type":"integer","value":75}`. + +Malformed, oversized, and unsupported-version requests receive one safe error +then their connection closes. Each connection has a bounded outbound queue; +newer unsent state events may replace older unsent events, while replies do not +coalesce. Queue overflow disconnects only that client. + +This initial local transport deliberately omits profiles, compound commands, +dangerous-control confirmation, and idempotency keys because service semantics +for those operations are not complete. Phase 4a profile work and Phase 5 LAN +API extend the protocol without changing daemon ownership or snapshot-resync +rules. + ## State rules - Snapshot includes `instance_id`, device identity, connection state, diff --git a/docs/roadmap.md b/docs/roadmap.md index cb72e8f..aff0d5b 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 — complete pending review and merge +## Phase 3: Pi compatibility verification — complete Execution plan: [Phase 3 Pi compatibility plan](phases/phase-3-pi-compatibility.md). @@ -52,9 +52,10 @@ 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 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. +MRs 1–4 are merged and confirmed. Native Pi evidence covers bounded +read-only discovery, event-driven reconciliation, physical unplug/replug, and +post-reboot daemon startup. GUI clients later cache state and cap rendering at +60 Hz. Before adding Phase 3 hardware coverage, split Solo tests into a read-only hardware suite (discovery, external changes, reconnect) and a write-capable @@ -66,6 +67,8 @@ deployment prerequisites are recorded. ## Phase 4a: Local touchscreen — planned +Execution plan: [Phase 4a local touchscreen plan](phases/phase-4a-local-touchscreen.md). + Implement versioned Unix-socket snapshot, command, and event messages, then build fullscreen Rust touch UI using only that local API. Start with main monitor/output controls; add capability groups after hardware and screen-fit From 2704106776f099445e842fc9f840175185d334a9 Mon Sep 17 00:00:00 2001 From: Tzu-Chieh Huang Date: Fri, 17 Jul 2026 20:19:00 -0700 Subject: [PATCH 2/3] fix(ipc): harden local client handling --- crates/focusrited/src/ipc.rs | 145 ++++++++++++++++++++++++++++++----- 1 file changed, 127 insertions(+), 18 deletions(-) diff --git a/crates/focusrited/src/ipc.rs b/crates/focusrited/src/ipc.rs index 86e9f0d..9b01769 100644 --- a/crates/focusrited/src/ipc.rs +++ b/crates/focusrited/src/ipc.rs @@ -76,7 +76,14 @@ impl Drop for LocalServer { fn remove_stale_socket(path: &Path) -> io::Result<()> { match fs::symlink_metadata(path) { - Ok(metadata) if metadata.file_type().is_socket() => fs::remove_file(path), + Ok(metadata) if metadata.file_type().is_socket() => match UnixStream::connect(path) { + Ok(_) => Err(io::Error::new( + io::ErrorKind::AddrInUse, + "socket is already owned by a running daemon", + )), + Err(error) if error.kind() == io::ErrorKind::ConnectionRefused => fs::remove_file(path), + Err(error) => Err(error), + }, Ok(_) => Err(io::Error::new( io::ErrorKind::AlreadyExists, "socket path exists and is not a socket", @@ -110,9 +117,12 @@ fn run(listener: UnixListener, worker: Arc, running: Arc, instance_id: &str) -> bool { + let mut remaining = REQUEST_BATCH_LIMIT; + if !process_requests(client, worker, instance_id, &mut remaining) { + return false; + } + if client.input.contains(&b'\n') { + return true; + } let mut bytes = [0; 4096]; match client.stream.read(&mut bytes) { Ok(0) => return false, @@ -157,10 +174,22 @@ fn read_client(client: &mut Client, worker: &Arc, instance_id: &st Err(error) if error.kind() == io::ErrorKind::WouldBlock => {} Err(_) => return false, } - for _ in 0..REQUEST_BATCH_LIMIT { + process_requests(client, worker, instance_id, &mut remaining) +} + +fn process_requests( + client: &mut Client, + worker: &Arc, + instance_id: &str, + remaining: &mut usize, +) -> bool { + while *remaining > 0 { let Some(end) = client.input.iter().position(|byte| *byte == b'\n') else { break; }; + if end + 1 > MAX_MESSAGE_BYTES { + return client.reject("message_too_large"); + } let line = client.input.drain(..=end).collect::>(); let line = &line[..line.len() - 1]; match serde_json::from_slice::(line) { @@ -176,6 +205,7 @@ fn read_client(client: &mut Client, worker: &Arc, instance_id: &st return client.reject("malformed_message"); } } + *remaining -= 1; } true } @@ -200,30 +230,46 @@ fn handle_request( }, _ => error_message("unknown_message_type"), }; - client.queue(Outbound::Reply(message)) + if client.queue(Outbound::Reply(message)) { + true + } else { + client.closing = true; + true + } } fn flush_client(client: &mut Client) -> bool { - while let Some(message) = client.outbound.front_mut() { - match client.stream.write(&message.bytes[message.offset..]) { - Ok(0) => return false, - Ok(count) => message.offset += count, - Err(error) if error.kind() == io::ErrorKind::WouldBlock => return true, - Err(_) => return false, + loop { + if let Some(message) = client.outbound.front_mut() { + match client.stream.write(&message.bytes[message.offset..]) { + Ok(0) => return false, + Ok(count) => message.offset += count, + Err(error) if error.kind() == io::ErrorKind::WouldBlock => return true, + Err(_) => return false, + } + if let Some(message) = client + .outbound + .pop_front_if(|message| message.offset == message.bytes.len()) + { + client.outbound_bytes -= message.bytes.len(); + } + continue; } - if let Some(message) = client - .outbound - .pop_front_if(|message| message.offset == message.bytes.len()) - { - client.outbound_bytes -= message.bytes.len(); + if let Some(error) = client.rejection.take() { + if !client.queue(Outbound::Reply(error_message(error))) { + return false; + } + continue; } + return true; } - true } fn broadcast(clients: &mut [Client], message: Outbound) { for client in clients { - let _ = client.queue(message.clone()); + if !client.queue(message.clone()) { + client.closing = true; + } } } @@ -233,6 +279,7 @@ struct Client { outbound: VecDeque, outbound_bytes: usize, closing: bool, + rejection: Option<&'static str>, } impl Client { @@ -243,6 +290,7 @@ impl Client { outbound: VecDeque::new(), outbound_bytes: 0, closing: false, + rejection: None, } } @@ -283,7 +331,12 @@ impl Client { fn reject(&mut self, error: &'static str) -> bool { self.closing = true; - self.queue(Outbound::Reply(error_message(error))) + self.rejection = Some(error); + true + } + + fn has_pending_output(&self) -> bool { + !self.outbound.is_empty() || self.rejection.is_some() } } @@ -627,6 +680,52 @@ mod tests { }; while client.queue(large_reply()) {} assert!(client.outbound_bytes <= MAX_OUTBOUND_BYTES); + + let (stream, _peer) = UnixStream::pair().unwrap(); + let mut full_client = Client::new(stream); + full_client.outbound.push_back(Message { + bytes: vec![b'x'; MAX_OUTBOUND_BYTES], + offset: 0, + event: false, + }); + full_client.outbound_bytes = MAX_OUTBOUND_BYTES; + let mut clients = vec![full_client]; + broadcast(&mut clients, event(3)); + assert!(clients[0].closing); + } + + #[test] + fn valid_pipelined_requests_are_limited_per_message() { + let (server, worker, _) = server(); + let (stream, _peer) = UnixStream::pair().unwrap(); + let mut client = Client::new(stream); + client.input = b"{\"v\":1,\"type\":\"snapshot\"}\n".repeat(3_000); + + assert!(read_client(&mut client, &worker, "test")); + assert!(!client.closing); + assert!(client.input.len() > MAX_MESSAGE_BYTES); + + server.stop(); + worker.stop().unwrap(); + } + + #[test] + fn rejection_waits_for_queued_output_before_disconnect() { + let (stream, _peer) = UnixStream::pair().unwrap(); + let mut client = Client::new(stream); + assert!(client.queue(Outbound::Reply(Response { + v: PROTOCOL_VERSION, + kind: "snapshot", + instance_id: None, + revision: None, + online: None, + snapshot: None, + error: None, + }))); + assert!(client.reject("malformed_message")); + assert!(client.has_pending_output()); + assert!(flush_client(&mut client)); + assert!(!client.has_pending_output()); } #[test] @@ -648,4 +747,14 @@ mod tests { server.stop(); worker.stop().unwrap(); } + + #[test] + fn startup_keeps_live_daemon_socket_intact() { + let (server, worker, path) = server(); + assert!(LocalServer::start(Arc::clone(&worker), path.clone()).is_err()); + assert!(UnixStream::connect(&path).is_ok()); + + server.stop(); + worker.stop().unwrap(); + } } From 4047bbc793935d50ebd548bdfd70c569f23b421d Mon Sep 17 00:00:00 2001 From: Tzu-Chieh Huang Date: Fri, 17 Jul 2026 20:27:57 -0700 Subject: [PATCH 3/3] fix(ipc): stop unsafe queue draining --- crates/focusrited/src/ipc.rs | 73 ++++++++++++++++++++++++++++++++++-- 1 file changed, 69 insertions(+), 4 deletions(-) diff --git a/crates/focusrited/src/ipc.rs b/crates/focusrited/src/ipc.rs index 9b01769..264cc66 100644 --- a/crates/focusrited/src/ipc.rs +++ b/crates/focusrited/src/ipc.rs @@ -162,19 +162,25 @@ fn read_client(client: &mut Client, worker: &Arc, instance_id: &st if client.input.contains(&b'\n') { return true; } + if client.input.len() > MAX_MESSAGE_BYTES { + return client.reject("message_too_large"); + } let mut bytes = [0; 4096]; match client.stream.read(&mut bytes) { Ok(0) => return false, Ok(count) => { client.input.extend_from_slice(&bytes[..count]); - if client.input.len() > MAX_MESSAGE_BYTES { - return client.reject("message_too_large"); - } } Err(error) if error.kind() == io::ErrorKind::WouldBlock => {} Err(_) => return false, } - process_requests(client, worker, instance_id, &mut remaining) + if !process_requests(client, worker, instance_id, &mut remaining) { + return false; + } + if !client.input.contains(&b'\n') && client.input.len() > MAX_MESSAGE_BYTES { + return client.reject("message_too_large"); + } + true } fn process_requests( @@ -206,6 +212,9 @@ fn process_requests( } } *remaining -= 1; + if client.closing { + return true; + } } true } @@ -267,6 +276,9 @@ fn flush_client(client: &mut Client) -> bool { fn broadcast(clients: &mut [Client], message: Outbound) { for client in clients { + if client.closing { + continue; + } if !client.queue(message.clone()) { client.closing = true; } @@ -692,6 +704,34 @@ mod tests { let mut clients = vec![full_client]; broadcast(&mut clients, event(3)); assert!(clients[0].closing); + let queued = clients[0].outbound_bytes; + broadcast(&mut clients, event(4)); + assert_eq!(clients[0].outbound_bytes, queued); + } + + #[test] + fn reply_overflow_stops_remaining_commands() { + let (server, worker, _) = server(); + let (stream, _peer) = UnixStream::pair().unwrap(); + let mut client = Client::new(stream); + client.outbound_bytes = MAX_OUTBOUND_BYTES; + client.input = b"{\"v\":1,\"type\":\"command\",\"control\":\"output.level\",\"value\":{\"type\":\"integer\",\"value\":60}}\n{\"v\":1,\"type\":\"command\",\"control\":\"output.level\",\"value\":{\"type\":\"integer\",\"value\":80}}\n".to_vec(); + let mut remaining = REQUEST_BATCH_LIMIT; + + assert!(process_requests( + &mut client, + &worker, + "test", + &mut remaining + )); + assert!(client.closing); + assert_eq!( + worker.state().unwrap().snapshot.values[&ControlId("output.level".into())], + Value::Integer(60) + ); + + server.stop(); + worker.stop().unwrap(); } #[test] @@ -709,6 +749,31 @@ mod tests { worker.stop().unwrap(); } + #[test] + fn completed_frame_does_not_count_against_next_partial_frame() { + let (server, worker, _) = server(); + let (stream, mut peer) = UnixStream::pair().unwrap(); + stream.set_nonblocking(true).unwrap(); + let mut client = Client::new(stream); + let prefix = "{\"v\":1,\"type\":\"snapshot\",\"padding\":\""; + let suffix = "\"}"; + let frame = format!( + "{prefix}{}{suffix}", + "x".repeat(MAX_MESSAGE_BYTES - 1 - prefix.len() - suffix.len()) + ); + let split = frame.len() - 4; + client.input = frame.as_bytes()[..split].to_vec(); + peer.write_all(&frame.as_bytes()[split..]).unwrap(); + peer.write_all(b"\n{\"v\":1,\"type\":\"snapshot\"").unwrap(); + + assert!(read_client(&mut client, &worker, "test")); + assert!(!client.closing); + assert!(client.input.len() < MAX_MESSAGE_BYTES); + + server.stop(); + worker.stop().unwrap(); + } + #[test] fn rejection_waits_for_queued_output_before_disconnect() { let (stream, _peer) = UnixStream::pair().unwrap();