From 16edecd129dd4e1cdbcf6a4d3c16a5fd3a52ab76 Mon Sep 17 00:00:00 2001 From: zackees Date: Mon, 29 Jun 2026 19:07:36 -0700 Subject: [PATCH 1/4] wip: fix serial monitor ack ordering --- crates/fbuild-cli/src/cli/serial_probe.rs | 24 +-- .../fbuild-python/src/async_serial_monitor.rs | 37 +++- crates/fbuild-python/src/json_rpc.rs | 64 +++++- crates/fbuild-python/src/serial_monitor.rs | 191 ++++++++++++------ crates/fbuild-serial/src/boards.rs | 44 ++++ crates/fbuild-serial/src/manager.rs | 58 ++++-- 6 files changed, 291 insertions(+), 127 deletions(-) diff --git a/crates/fbuild-cli/src/cli/serial_probe.rs b/crates/fbuild-cli/src/cli/serial_probe.rs index 6deac62f..7d65745d 100644 --- a/crates/fbuild-cli/src/cli/serial_probe.rs +++ b/crates/fbuild-cli/src/cli/serial_probe.rs @@ -35,7 +35,7 @@ use std::time::{Duration, Instant}; use clap::Subcommand; use fbuild_core::{FbuildError, Result}; -use fbuild_serial::boards::{board_hint, family_for_vid_pid, vcom_for_env, BoardFamily}; +use fbuild_serial::boards::{board_hint, family_for_port_or_default, vcom_for_env}; use crate::output; @@ -222,7 +222,7 @@ fn read_port(port_name: &str, baud: u32, seconds: f64, send: Option<&str>) -> Re // if `serialport` can see it. Unknown VID:PID → safe-default // `CdcAcmBridge` (DTR=true, RTS=true) per the FastLED/FastLED#3300 // bug mode — better to over-assert than to silently drop. - let family = infer_family_for_port(port_name); + let family = family_for_port_or_default(port_name); let (dtr, rts) = family.idle_dtr_rts(); let mut handle = serialport::new(port_name, baud) @@ -268,26 +268,6 @@ fn read_port(port_name: &str, baud: u32, seconds: f64, send: Option<&str>) -> Re Ok(()) } -/// Walk `serialport::available_ports()` once and return the family of -/// the port that matches `name`. Fall through to the safe-default -/// CDC-ACM convention if the port isn't enumerable OR the VID/PID -/// is unknown. -fn infer_family_for_port(name: &str) -> BoardFamily { - if let Ok(ports) = serialport::available_ports() { - for port in ports { - if port.port_name != name { - continue; - } - if let serialport::SerialPortType::UsbPort(info) = port.port_type { - if let Some(family) = family_for_vid_pid(info.vid, info.pid) { - return family; - } - } - } - } - BoardFamily::CdcAcmBridge -} - /// Interpret a handful of backslash escapes in `--send` payloads so /// scripted probes (`--send 'hello\n'`, `--send '\x7e'`) work without /// shell quoting gymnastics. Anything we don't recognize passes diff --git a/crates/fbuild-python/src/async_serial_monitor.rs b/crates/fbuild-python/src/async_serial_monitor.rs index f39f3d79..cb756aff 100644 --- a/crates/fbuild-python/src/async_serial_monitor.rs +++ b/crates/fbuild-python/src/async_serial_monitor.rs @@ -4,6 +4,7 @@ use futures::{SinkExt, StreamExt}; use pyo3::prelude::*; use serde::Serialize; +use std::collections::VecDeque; use std::sync::Arc; use tokio_tungstenite::tungstenite; @@ -62,6 +63,7 @@ pub(crate) struct AsyncSerialMonitor { client_id: String, ws_write: Arc>>, ws_read: Arc>>, + pending_lines: Arc>>, } #[pymethods] @@ -77,6 +79,7 @@ impl AsyncSerialMonitor { client_id: uuid::Uuid::new_v4().to_string(), ws_write: Arc::new(tokio::sync::Mutex::new(None)), ws_read: Arc::new(tokio::sync::Mutex::new(None)), + pending_lines: Arc::new(tokio::sync::Mutex::new(VecDeque::new())), } } @@ -91,6 +94,7 @@ impl AsyncSerialMonitor { let verbose = slf.verbose; let ws_write_slot = slf.ws_write.clone(); let ws_read_slot = slf.ws_read.clone(); + let pending_lines = slf.pending_lines.clone(); let slf_obj: PyObject = slf.into_py(py); pyo3_async_runtimes::tokio::future_into_py(py, async move { @@ -200,6 +204,7 @@ impl AsyncSerialMonitor { *ws_write_slot.lock().await = Some(write); *ws_read_slot.lock().await = Some(read); + pending_lines.lock().await.clear(); Ok(slf_obj) }) @@ -217,6 +222,7 @@ impl AsyncSerialMonitor { ) -> PyResult> { let ws_write_slot = self.ws_write.clone(); let ws_read_slot = self.ws_read.clone(); + let pending_lines = self.pending_lines.clone(); pyo3_async_runtimes::tokio::future_into_py(py, async move { if let Some(mut write) = ws_write_slot.lock().await.take() { @@ -226,6 +232,7 @@ impl AsyncSerialMonitor { let _ = write.send(tungstenite::Message::Close(None)).await; } let _ = ws_read_slot.lock().await.take(); + pending_lines.lock().await.clear(); Ok(false) }) } @@ -237,10 +244,11 @@ impl AsyncSerialMonitor { #[pyo3(signature = (timeout_secs=30.0))] fn read_lines<'py>(&self, py: Python<'py>, timeout_secs: f64) -> PyResult> { let ws_read_slot = self.ws_read.clone(); + let pending_lines = self.pending_lines.clone(); let auto_reconnect = self.auto_reconnect; pyo3_async_runtimes::tokio::future_into_py(py, async move { - Ok(read_lines_async(ws_read_slot, auto_reconnect, timeout_secs).await) + Ok(read_lines_async(ws_read_slot, pending_lines, auto_reconnect, timeout_secs).await) }) } @@ -252,10 +260,11 @@ impl AsyncSerialMonitor { fn write<'py>(&self, py: Python<'py>, data: &str) -> PyResult> { let ws_write_slot = self.ws_write.clone(); let ws_read_slot = self.ws_read.clone(); + let pending_lines = self.pending_lines.clone(); let encoded = encode_payload(data); pyo3_async_runtimes::tokio::future_into_py(py, async move { - Ok(write_async(ws_write_slot, ws_read_slot, encoded).await) + Ok(write_async(ws_write_slot, ws_read_slot, pending_lines, encoded).await) }) } @@ -287,6 +296,7 @@ impl AsyncSerialMonitor { let ws_write_slot = self.ws_write.clone(); let ws_read_slot = self.ws_read.clone(); + let pending_lines = self.pending_lines.clone(); let auto_reconnect = self.auto_reconnect; pyo3_async_runtimes::tokio::future_into_py(py, async move { @@ -294,11 +304,21 @@ impl AsyncSerialMonitor { // a write failure because the sync surface also proceeds // to poll for a REMOTE: response (which is how the Python // tests exercise this path). - let _ = write_async(ws_write_slot, ws_read_slot.clone(), encoded).await; + let _ = write_async( + ws_write_slot, + ws_read_slot.clone(), + pending_lines.clone(), + encoded, + ) + .await; - let json_part = - wait_for_remote_json_rpc_response_async(timeout_secs, ws_read_slot, auto_reconnect) - .await; + let json_part = wait_for_remote_json_rpc_response_async( + timeout_secs, + ws_read_slot, + pending_lines, + auto_reconnect, + ) + .await; match json_part { Some(payload) => Python::with_gil(|py| { @@ -361,10 +381,7 @@ pub(crate) async fn post_reset_request_async( })?; let body: serde_json::Value = resp.json().await.map_err(|e| { - pyo3::exceptions::PyRuntimeError::new_err(format!( - "failed to parse reset response: {}", - e - )) + pyo3::exceptions::PyRuntimeError::new_err(format!("failed to parse reset response: {}", e)) })?; Ok(body diff --git a/crates/fbuild-python/src/json_rpc.rs b/crates/fbuild-python/src/json_rpc.rs index 7b0c4b9c..fe16ddbc 100644 --- a/crates/fbuild-python/src/json_rpc.rs +++ b/crates/fbuild-python/src/json_rpc.rs @@ -3,6 +3,7 @@ use base64::Engine; use futures::{SinkExt, StreamExt}; +use std::collections::VecDeque; use std::sync::Arc; use tokio_tungstenite::tungstenite; @@ -38,10 +39,21 @@ where /// futures can still progress between iterations. pub(crate) async fn read_lines_async( ws_read_slot: Arc>>, + pending_lines: Arc>>, auto_reconnect: bool, timeout_secs: f64, ) -> Vec { let mut lines = Vec::new(); + { + let mut pending = pending_lines.lock().await; + while let Some(line) = pending.pop_front() { + lines.push(line); + } + } + if !lines.is_empty() { + return lines; + } + let deadline = std::time::Instant::now() + std::time::Duration::from_secs_f64(timeout_secs); loop { @@ -103,6 +115,7 @@ pub(crate) async fn read_lines_async( pub(crate) async fn write_async( ws_write_slot: Arc>>, ws_read_slot: Arc>>, + pending_lines: Arc>>, encoded: String, ) -> bool { let msg = serde_json::to_string(&ClientMessage::Write { data: encoded }) @@ -118,22 +131,44 @@ pub(crate) async fn write_async( } } - // Wait for write_ack. Hold the read half's mutex only across this - // single `.next()` so concurrent `read_lines` futures can resume. + // Wait for write_ack. Serial data can arrive before the ack on the + // WebSocket; queue it for the next read rather than consuming it. let mut guard = ws_read_slot.lock().await; let Some(source) = guard.as_mut() else { return false; }; - let ack_timeout = std::time::Duration::from_secs(5); - match tokio::time::timeout(ack_timeout, source.next()).await { - Ok(Some(Ok(tungstenite::Message::Text(text)))) => { - matches!( - serde_json::from_str::(&text), - Ok(ServerMessage::WriteAck { bytes_written, .. }) if bytes_written > 0 - ) + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); + while std::time::Instant::now() < deadline { + let remaining = deadline - std::time::Instant::now(); + match tokio::time::timeout(remaining, source.next()).await { + Ok(Some(Ok(tungstenite::Message::Text(text)))) => { + match serde_json::from_str::(&text) { + Ok(ServerMessage::WriteAck { + success, + bytes_written, + .. + }) => return success && bytes_written > 0, + Ok(ServerMessage::Data { lines, .. }) => { + pending_lines.lock().await.extend(lines); + continue; + } + Ok(ServerMessage::Preempted { .. }) + | Ok(ServerMessage::Reconnected { .. }) + | Ok(ServerMessage::PortRenumbered { .. }) + | Ok(ServerMessage::PortReattached { .. }) + | Ok(ServerMessage::Other) => continue, + Ok(ServerMessage::Error { .. }) + | Ok(ServerMessage::PortDisconnected { .. }) + | Ok(ServerMessage::PortRebindFailed { .. }) => return false, + _ => continue, + } + } + Ok(Some(Ok(tungstenite::Message::Close(_)))) | Ok(None) => break, + Err(_) => break, + _ => continue, } - _ => false, } + false } /// Async counterpart to `wait_for_remote_json_rpc_response`. Keeps @@ -142,6 +177,7 @@ pub(crate) async fn write_async( pub(crate) async fn wait_for_remote_json_rpc_response_async( timeout_secs: f64, ws_read_slot: Arc>>, + pending_lines: Arc>>, auto_reconnect: bool, ) -> Option { let deadline = std::time::Instant::now() + std::time::Duration::from_secs_f64(timeout_secs); @@ -151,7 +187,13 @@ pub(crate) async fn wait_for_remote_json_rpc_response_async( if remaining <= 0.0 { break; } - let lines = read_lines_async(ws_read_slot.clone(), auto_reconnect, remaining).await; + let lines = read_lines_async( + ws_read_slot.clone(), + pending_lines.clone(), + auto_reconnect, + remaining, + ) + .await; if let Some(json_part) = extract_remote_json_rpc_response(&lines) { return Some(json_part); } diff --git a/crates/fbuild-python/src/serial_monitor.rs b/crates/fbuild-python/src/serial_monitor.rs index 5ca487c6..46c2d65d 100644 --- a/crates/fbuild-python/src/serial_monitor.rs +++ b/crates/fbuild-python/src/serial_monitor.rs @@ -3,6 +3,7 @@ use base64::Engine; use futures::{SinkExt, StreamExt}; use pyo3::prelude::*; +use std::collections::VecDeque; use std::sync::Mutex; use tokio::runtime::Runtime; use tokio_tungstenite::tungstenite; @@ -38,6 +39,7 @@ pub(crate) struct SerialMonitor { runtime: Option<&'static Runtime>, ws_write: Option>, ws_read: Option>, + pending_lines: Mutex>, client_id: String, last_line: String, #[allow(dead_code)] @@ -56,8 +58,7 @@ impl SerialMonitor { const HANDSHAKE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5); let connect_result = rt.block_on(async { - tokio::time::timeout(HANDSHAKE_TIMEOUT, tokio_tungstenite::connect_async(&ws_url)) - .await + tokio::time::timeout(HANDSHAKE_TIMEOUT, tokio_tungstenite::connect_async(&ws_url)).await }); let (ws_stream, _) = match connect_result { Ok(Ok(ok)) => ok, @@ -88,8 +89,11 @@ impl SerialMonitor { .expect("fbuild-python: ClientMessage::Attach serialization is infallible"); let send_result = rt.block_on(async { - tokio::time::timeout(HANDSHAKE_TIMEOUT, write.send(tungstenite::Message::Text(attach_json))) - .await + tokio::time::timeout( + HANDSHAKE_TIMEOUT, + write.send(tungstenite::Message::Text(attach_json)), + ) + .await }); match send_result { Ok(Ok(())) => {} @@ -106,9 +110,8 @@ impl SerialMonitor { } } - let read_result = rt.block_on(async { - tokio::time::timeout(HANDSHAKE_TIMEOUT, read.next()).await - }); + let read_result = + rt.block_on(async { tokio::time::timeout(HANDSHAKE_TIMEOUT, read.next()).await }); let msg: tungstenite::Message = match read_result { Ok(Some(Ok(msg))) => msg, Ok(Some(Err(e))) => { @@ -164,6 +167,7 @@ impl SerialMonitor { } self.ws_write = None; self.ws_read = None; + self.clear_pending_lines(); } fn reconnect_ws(&mut self) -> PyResult<()> { @@ -175,6 +179,35 @@ impl SerialMonitor { self.ws_read = Some(Mutex::new(read)); Ok(()) } + + fn push_pending_lines(&self, lines: Vec) { + if lines.is_empty() { + return; + } + let mut pending = self.pending_lines.lock().unwrap_or_else(|e| e.into_inner()); + pending.extend(lines); + } + + fn drain_pending_lines_into(&self, lines: &mut Vec) { + let mut pending = self.pending_lines.lock().unwrap_or_else(|e| e.into_inner()); + while let Some(line) = pending.pop_front() { + lines.push(line); + } + } + + fn pending_line_count(&self) -> usize { + self.pending_lines + .lock() + .unwrap_or_else(|e| e.into_inner()) + .len() + } + + fn clear_pending_lines(&self) { + self.pending_lines + .lock() + .unwrap_or_else(|e| e.into_inner()) + .clear(); + } } #[pymethods] @@ -197,6 +230,7 @@ impl SerialMonitor { runtime: None, ws_write: None, ws_read: None, + pending_lines: Mutex::new(VecDeque::new()), client_id: uuid::Uuid::new_v4().to_string(), last_line: String::new(), preempted: false, @@ -217,6 +251,7 @@ impl SerialMonitor { let (write, read) = slf.connect_ws(rt)?; slf.ws_write = Some(Mutex::new(write)); slf.ws_read = Some(Mutex::new(read)); + slf.clear_pending_lines(); slf.runtime = Some(rt); Ok(slf) } @@ -249,56 +284,59 @@ impl SerialMonitor { }; let mut lines = Vec::new(); + self.drain_pending_lines_into(&mut lines); let deadline = std::time::Instant::now() + std::time::Duration::from_secs_f64(timeout); let auto_reconnect = self.auto_reconnect; - py.allow_threads(|| { - while std::time::Instant::now() < deadline { - let remaining = deadline - std::time::Instant::now(); - let result = { - let mut read = ws_read.lock().unwrap_or_else(|e| e.into_inner()); - // tokio::time::timeout MUST be constructed inside the - // runtime context, otherwise it panics with "there is - // no reactor running" because the Sleep future needs - // Handle::current() to register with the timer driver. - rt.block_on(async { tokio::time::timeout(remaining, read.next()).await }) - }; - - match result { - Ok(Some(Ok(tungstenite::Message::Text(text)))) => { - match serde_json::from_str::(&text) { - Ok(ServerMessage::Data { - lines: data_lines, .. - }) => { - lines.extend(data_lines); - if !lines.is_empty() { + if lines.is_empty() { + py.allow_threads(|| { + while std::time::Instant::now() < deadline { + let remaining = deadline - std::time::Instant::now(); + let result = { + let mut read = ws_read.lock().unwrap_or_else(|e| e.into_inner()); + // tokio::time::timeout MUST be constructed inside the + // runtime context, otherwise it panics with "there is + // no reactor running" because the Sleep future needs + // Handle::current() to register with the timer driver. + rt.block_on(async { tokio::time::timeout(remaining, read.next()).await }) + }; + + match result { + Ok(Some(Ok(tungstenite::Message::Text(text)))) => { + match serde_json::from_str::(&text) { + Ok(ServerMessage::Data { + lines: data_lines, .. + }) => { + lines.extend(data_lines); + if !lines.is_empty() { + break; + } + } + Ok(ServerMessage::Preempted { .. }) => { + // Pause — deploy is happening + if auto_reconnect { + continue; + } break; } - } - Ok(ServerMessage::Preempted { .. }) => { - // Pause — deploy is happening - if auto_reconnect { + Ok(ServerMessage::Reconnected { .. }) => { + // Resume after deploy continue; } - break; + Ok(ServerMessage::PortRenumbered { .. }) + | Ok(ServerMessage::PortReattached { .. }) => continue, + Ok(ServerMessage::PortRebindFailed { .. }) => break, + Ok(ServerMessage::PortDisconnected { .. }) => break, + _ => continue, } - Ok(ServerMessage::Reconnected { .. }) => { - // Resume after deploy - continue; - } - Ok(ServerMessage::PortRenumbered { .. }) - | Ok(ServerMessage::PortReattached { .. }) => continue, - Ok(ServerMessage::PortRebindFailed { .. }) => break, - Ok(ServerMessage::PortDisconnected { .. }) => break, - _ => continue, } + Ok(Some(Ok(tungstenite::Message::Close(_)))) | Ok(None) => break, + Err(_) => break, // timeout + _ => continue, } - Ok(Some(Ok(tungstenite::Message::Close(_)))) | Ok(None) => break, - Err(_) => break, // timeout - _ => continue, } - } - }); + }); + } // Update last_line and dispatch hooks if let Some(last) = lines.last() { @@ -341,21 +379,42 @@ impl SerialMonitor { } } - // Wait for write_ack + // Wait for write_ack. Serial data can race ahead of the ack on the + // WebSocket; preserve it for the next read instead of discarding it. let mut read = ws_read.lock().unwrap_or_else(|e| e.into_inner()); - let timeout = std::time::Duration::from_secs(5); - // tokio::time::timeout must be created inside the runtime context. - match rt.block_on(async { tokio::time::timeout(timeout, read.next()).await }) { - Ok(Some(Ok(tungstenite::Message::Text(text)))) => { - if let Ok(ServerMessage::WriteAck { bytes_written, .. }) = - serde_json::from_str(&text) - { - return bytes_written; + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); + while std::time::Instant::now() < deadline { + let remaining = deadline - std::time::Instant::now(); + // tokio::time::timeout must be created inside the runtime context. + match rt.block_on(async { tokio::time::timeout(remaining, read.next()).await }) { + Ok(Some(Ok(tungstenite::Message::Text(text)))) => { + match serde_json::from_str::(&text) { + Ok(ServerMessage::WriteAck { + success, + bytes_written, + .. + }) => return if success { bytes_written } else { 0 }, + Ok(ServerMessage::Data { lines, .. }) => { + self.push_pending_lines(lines); + continue; + } + Ok(ServerMessage::Preempted { .. }) + | Ok(ServerMessage::Reconnected { .. }) + | Ok(ServerMessage::PortRenumbered { .. }) + | Ok(ServerMessage::PortReattached { .. }) + | Ok(ServerMessage::Other) => continue, + Ok(ServerMessage::Error { .. }) + | Ok(ServerMessage::PortDisconnected { .. }) + | Ok(ServerMessage::PortRebindFailed { .. }) => return 0, + _ => continue, + } } - 0 + Ok(Some(Ok(tungstenite::Message::Close(_)))) | Ok(None) => break, + Err(_) => break, + _ => continue, } - _ => 0, } + 0 } /// Run monitor until condition returns True or timeout expires. @@ -460,12 +519,16 @@ impl SerialMonitor { let result = rt.block_on(async { tokio::time::timeout(remaining, read.next()).await }); match result { Ok(Some(Ok(tungstenite::Message::Text(text)))) => { - if let Ok(ServerMessage::InWaiting { count }) = - serde_json::from_str::(&text) - { - return count; + match serde_json::from_str::(&text) { + Ok(ServerMessage::InWaiting { count }) => { + return self.pending_line_count() + count; + } + Ok(ServerMessage::Data { lines, .. }) => { + self.push_pending_lines(lines); + continue; + } + _ => continue, } - continue; } Ok(Some(Ok(tungstenite::Message::Close(_)))) | Ok(None) => break, Err(_) => break, @@ -482,6 +545,7 @@ impl SerialMonitor { /// FastLED/fbuild#605 — added as part of the deprecation of direct /// pyserial use by fbuild clients. fn reset_input_buffer(&self) { + self.clear_pending_lines(); let (Some(rt), Some(ws_write)) = (&self.runtime, &self.ws_write) else { return; }; @@ -598,10 +662,11 @@ impl SerialMonitor { }; let mut lines = Vec::new(); + self.drain_pending_lines_into(&mut lines); let deadline = std::time::Instant::now() + std::time::Duration::from_secs_f64(timeout); let auto_reconnect = self.auto_reconnect; - while std::time::Instant::now() < deadline { + while lines.is_empty() && std::time::Instant::now() < deadline { let remaining = deadline - std::time::Instant::now(); let result = { let mut read = ws_read.lock().unwrap_or_else(|e| e.into_inner()); diff --git a/crates/fbuild-serial/src/boards.rs b/crates/fbuild-serial/src/boards.rs index 023ab0e9..54f5fc43 100644 --- a/crates/fbuild-serial/src/boards.rs +++ b/crates/fbuild-serial/src/boards.rs @@ -489,6 +489,42 @@ pub fn family_for_vid_pid(vid: u16, pid: u16) -> Option { } } +/// Walk `serialport::available_ports()` once and classify the port that +/// matches `name`. +/// +/// Returns `None` when the port is not enumerable, is not a USB serial +/// port, or has an unknown VID/PID. Callers that need an idle DTR/RTS +/// state should prefer [`family_for_port_or_default`] so unknown +/// hardware keeps the CDC-ACM host-ready convention. +#[must_use] +pub fn family_for_port(name: &str) -> Option { + let ports = serialport::available_ports().ok()?; + for port in ports { + if !serial_port_name_matches(&port.port_name, name) { + continue; + } + if let serialport::SerialPortType::UsbPort(info) = port.port_type { + return family_for_vid_pid(info.vid, info.pid); + } + } + None +} + +/// Classify a serial port, falling back to the safe CDC-ACM host-ready +/// convention when the OS cannot report a known VID/PID. +#[must_use] +pub fn family_for_port_or_default(name: &str) -> BoardFamily { + family_for_port(name).unwrap_or(BoardFamily::CdcAcmBridge) +} + +fn serial_port_name_matches(candidate: &str, requested: &str) -> bool { + if cfg!(windows) { + candidate.eq_ignore_ascii_case(requested) + } else { + candidate == requested + } +} + #[cfg(test)] mod tests { use super::*; @@ -617,6 +653,14 @@ mod tests { assert_eq!(family_for_vid_pid(0, 0), None); } + #[test] + fn family_for_port_default_is_host_ready_cdc() { + assert_eq!( + family_for_port_or_default("__fbuild_missing_test_port__"), + BoardFamily::CdcAcmBridge + ); + } + /// The whole point of #684 + #686: any path that ends in /// `idle_dtr_rts()` getting `(false, false)` on a CDC-ACM bridge /// reintroduces the FastLED/FastLED#3300 silent-byte-drop bug. diff --git a/crates/fbuild-serial/src/manager.rs b/crates/fbuild-serial/src/manager.rs index 3958a8e4..87159ddd 100644 --- a/crates/fbuild-serial/src/manager.rs +++ b/crates/fbuild-serial/src/manager.rs @@ -66,6 +66,10 @@ impl SharedSerialManager { /// caller knows the chip is a native ESP USB CDC AND wants the /// "(false, false) = run firmware" post-open idle. Pass /// `Some(BoardFamily::Esp32NativeUsbCdc)` in that case. + /// + /// Daemon callers usually pass `None`; that path now infers native + /// ESP USB CDC from the OS-reported VID/PID before applying the + /// `(true, true)` unknown-port fallback. pub async fn open_port( &self, port: &str, @@ -106,11 +110,13 @@ impl SharedSerialManager { // self-eviction tick, HTTP handlers) keep making progress. See // ISSUES.md "Issue C". let port_for_open = port_name.clone(); - let family_for_open = family; + let explicit_family = family; let open_result: std::result::Result< std::result::Result, serialport::Error>, tokio::task::JoinError, > = tokio::task::spawn_blocking(move || { + let family_for_open = + explicit_family.or_else(|| crate::boards::family_for_port(&port_for_open)); let mut serial = serialport::new(&port_for_open, baud_rate) .timeout(Duration::from_millis(timeout_ms)) .open()?; @@ -131,7 +137,9 @@ impl SharedSerialManager { // incident (FastLED/FastLED#3300). For the full per-chip // matrix see `docs/usb-cdc-control-line-matrix.md` // (FastLED/fbuild#689). - let (dtr, rts) = family.map(|f| f.idle_dtr_rts()).unwrap_or((true, true)); + let (dtr, rts) = family_for_open + .map(|f| f.idle_dtr_rts()) + .unwrap_or((true, true)); match serial.write_data_terminal_ready(dtr) { Ok(()) => tracing::debug!( family = ?family_for_open, @@ -337,30 +345,28 @@ impl SharedSerialManager { let port_owned = port.to_string(); let data_owned = data.to_vec(); + let expected_len = data_owned.len(); let write_future = tokio::task::spawn_blocking(move || { use std::io::Write; let mut serial = handle.blocking_lock(); - let n = serial - .write(&data_owned) - .map_err(|e| format!("write failed: {}", e))?; serial - .flush() - .map_err(|e| format!("flush failed: {}", e))?; - Ok::(n) + .write_all(&data_owned) + .map_err(|e| format!("write failed: {}", e))?; + serial.flush().map_err(|e| format!("flush failed: {}", e))?; + Ok::(expected_len) }); // 2s budget: covers normal Windows USB-CDC drain + headroom but // bounds the worst case so a wedged adapter cannot stall the // daemon indefinitely on a write/flush syscall. - let join_result = - tokio::time::timeout(Duration::from_secs(2), write_future) - .await - .map_err(|_| { - fbuild_core::FbuildError::SerialError(format!( - "write to {} timed out after 2s", - port_owned - )) - })?; + let join_result = tokio::time::timeout(Duration::from_secs(2), write_future) + .await + .map_err(|_| { + fbuild_core::FbuildError::SerialError(format!( + "write to {} timed out after 2s", + port_owned + )) + })?; let bytes_written = match join_result { Ok(Ok(n)) => n, @@ -909,15 +915,25 @@ impl SharedSerialManager { std::result::Result, serialport::Error>, tokio::task::JoinError, > = tokio::task::spawn_blocking(move || { + let family_for_open = crate::boards::family_for_port(&port_for_open); + let (dtr, rts) = family_for_open + .map(|family| family.idle_dtr_rts()) + .unwrap_or((true, true)); let mut serial = serialport::new(&port_for_open, baud_rate) .timeout(Duration::from_millis(100)) .open()?; - match serial.write_data_terminal_ready(true) { - Ok(()) => tracing::debug!("manager: open-time DTR=high asserted"), + match serial.write_data_terminal_ready(dtr) { + Ok(()) => tracing::debug!( + family = ?family_for_open, + "manager: open-time DTR={dtr} asserted" + ), Err(e) => tracing::warn!("failed to set DTR: {}", e), } - match serial.write_request_to_send(true) { - Ok(()) => tracing::debug!("manager: open-time RTS=high asserted"), + match serial.write_request_to_send(rts) { + Ok(()) => tracing::debug!( + family = ?family_for_open, + "manager: open-time RTS={rts} asserted" + ), Err(e) => tracing::warn!("failed to set RTS: {}", e), } Ok(serial) From 5abbb1b61b82918469ee6bbc81be26c484ffd30b Mon Sep 17 00:00:00 2001 From: zackees Date: Tue, 30 Jun 2026 08:35:45 -0700 Subject: [PATCH 2/4] test(json-rpc): regression guards for write_ack vs data ordering MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two unit tests in fbuild-python covering the new pending_lines queue: - read_lines_async_drains_pending_before_wire — proves drain happens before any wire access (uses a None ws slot). - write_async_queues_data_arriving_before_ack — spins up an in-process tokio-tungstenite server that sends Data ahead of WriteAck and asserts the data lines land in pending_lines while write_async still returns true on the eventual ack. Co-Authored-By: Claude Opus 4.7 (1M context) --- crates/fbuild-python/src/lib.rs | 82 +++++++++++++++++++++++++++++++++ 1 file changed, 82 insertions(+) diff --git a/crates/fbuild-python/src/lib.rs b/crates/fbuild-python/src/lib.rs index 5f4666d5..cb1d74e9 100644 --- a/crates/fbuild-python/src/lib.rs +++ b/crates/fbuild-python/src/lib.rs @@ -546,4 +546,86 @@ mod tests { "src_dir must serialize verbatim when set, got {json}" ); } + + /// `read_lines_async` must drain the cross-call `pending_lines` queue + /// before touching the wire. The PR fix for write_ack ordering parks + /// Data frames that arrive ahead of the ack into this queue, so the + /// next `read_lines` MUST see them even if the wire is idle (or, as + /// here, never connected). + #[test] + fn read_lines_async_drains_pending_before_wire() { + use crate::json_rpc::read_lines_async; + use std::collections::VecDeque; + use std::sync::Arc; + let rt = tokio::runtime::Runtime::new().unwrap(); + rt.block_on(async { + let ws_read_slot = Arc::new(tokio::sync::Mutex::new(None)); + let pending = Arc::new(tokio::sync::Mutex::new(VecDeque::from(vec![ + "line-a".to_string(), + "line-b".to_string(), + ]))); + let lines = read_lines_async(ws_read_slot, pending.clone(), false, 0.1).await; + assert_eq!(lines, vec!["line-a".to_string(), "line-b".to_string()]); + assert!( + pending.lock().await.is_empty(), + "pending queue must be drained after read" + ); + }); + } + + /// Regression guard for the write_ack vs Data ordering bug: if a + /// `Data` frame lands on the wire before the `WriteAck`, `write_async` + /// must enqueue the data lines into `pending_lines` (so the next read + /// surfaces them) AND still return `true` once the ack arrives. + /// Previous behavior silently consumed the data frame. + #[test] + fn write_async_queues_data_arriving_before_ack() { + use crate::json_rpc::write_async; + use futures::{SinkExt, StreamExt}; + use std::collections::VecDeque; + use std::sync::Arc; + use tokio_tungstenite::tungstenite; + + let rt = tokio::runtime::Runtime::new().unwrap(); + rt.block_on(async { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let server = tokio::spawn(async move { + let (sock, _) = listener.accept().await.unwrap(); + let mut ws = tokio_tungstenite::accept_async(sock).await.unwrap(); + let _ = ws.next().await; + ws.send(tungstenite::Message::Text( + r#"{"type":"data","lines":["serial-out"],"current_index":1}"#.to_string(), + )) + .await + .unwrap(); + ws.send(tungstenite::Message::Text( + r#"{"type":"write_ack","success":true,"bytes_written":3,"message":null}"# + .to_string(), + )) + .await + .unwrap(); + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + }); + + let url = format!("ws://{}/", addr); + let (ws_stream, _) = tokio_tungstenite::connect_async(url).await.unwrap(); + let (sink, source) = ws_stream.split(); + + let ws_write = Arc::new(tokio::sync::Mutex::new(Some(sink))); + let ws_read = Arc::new(tokio::sync::Mutex::new(Some(source))); + let pending = Arc::new(tokio::sync::Mutex::new(VecDeque::::new())); + + let ok = write_async(ws_write, ws_read, pending.clone(), "AAA=".to_string()).await; + assert!(ok, "write_async must return true after seeing WriteAck"); + let pending_snapshot: Vec = pending.lock().await.iter().cloned().collect(); + assert_eq!( + pending_snapshot, + vec!["serial-out".to_string()], + "data frame arriving before write_ack must be queued for the next read" + ); + + let _ = server.await; + }); + } } From da3193b985f3fc0d3c1f933a62eb6cf8d6153a1e Mon Sep 17 00:00:00 2001 From: zackees Date: Tue, 30 Jun 2026 08:54:57 -0700 Subject: [PATCH 3/4] test(serial-manager): regression guard for write_all partial-write loop MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit PartialWriteSerialPort returns Ok(2) on the first write() call and full length thereafter — exactly the OS-level back-pressure pattern that broke the prior write()-then-return path. write_to_port must report the full payload length (write_all contract) and the OS-side buffer must contain every byte. Prevents regressing to a write_ack with a partial bytes_written count. Co-Authored-By: Claude Opus 4.7 (1M context) --- crates/fbuild-serial/src/manager/tests.rs | 181 ++++++++++++++++++++++ 1 file changed, 181 insertions(+) diff --git a/crates/fbuild-serial/src/manager/tests.rs b/crates/fbuild-serial/src/manager/tests.rs index d0bc187d..d7658a5a 100644 --- a/crates/fbuild-serial/src/manager/tests.rs +++ b/crates/fbuild-serial/src/manager/tests.rs @@ -568,3 +568,184 @@ async fn rebind_preserves_session_and_routes_writes_to_new_handle() { assert!(mgr.sessions.get(old_port).is_none()); assert!(mgr.port_aliases.get(new_port).is_none()); } + +/// Fake port whose first `write()` only accepts the first `partial_n` +/// bytes — exactly the OS-level behavior that breaks naive `write()` +/// callers. Subsequent writes accept everything offered. +#[derive(Clone)] +struct PartialWriteSerialPort { + name: String, + writes: Arc>>, + first_chunk: Arc>>, +} + +impl PartialWriteSerialPort { + fn new(name: &str, partial_n: usize) -> (Self, Arc>>) { + let writes = Arc::new(std::sync::Mutex::new(Vec::new())); + ( + Self { + name: name.to_string(), + writes: Arc::clone(&writes), + first_chunk: Arc::new(std::sync::Mutex::new(Some(partial_n))), + }, + writes, + ) + } +} + +impl Read for PartialWriteSerialPort { + fn read(&mut self, _buf: &mut [u8]) -> std::io::Result { + Err(std::io::Error::new(std::io::ErrorKind::TimedOut, "no data")) + } +} + +impl Write for PartialWriteSerialPort { + fn write(&mut self, buf: &[u8]) -> std::io::Result { + let take = { + let mut first = self.first_chunk.lock().unwrap(); + match first.take() { + Some(n) => n.min(buf.len()), + None => buf.len(), + } + }; + self.writes.lock().unwrap().extend_from_slice(&buf[..take]); + Ok(take) + } + + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } +} + +impl serialport::SerialPort for PartialWriteSerialPort { + fn name(&self) -> Option { + Some(self.name.clone()) + } + fn baud_rate(&self) -> serialport::Result { + Ok(115200) + } + fn data_bits(&self) -> serialport::Result { + Ok(DataBits::Eight) + } + fn flow_control(&self) -> serialport::Result { + Ok(FlowControl::None) + } + fn parity(&self) -> serialport::Result { + Ok(Parity::None) + } + fn stop_bits(&self) -> serialport::Result { + Ok(StopBits::One) + } + fn timeout(&self) -> Duration { + Duration::from_millis(100) + } + fn set_baud_rate(&mut self, _: u32) -> serialport::Result<()> { + Ok(()) + } + fn set_data_bits(&mut self, _: DataBits) -> serialport::Result<()> { + Ok(()) + } + fn set_flow_control(&mut self, _: FlowControl) -> serialport::Result<()> { + Ok(()) + } + fn set_parity(&mut self, _: Parity) -> serialport::Result<()> { + Ok(()) + } + fn set_stop_bits(&mut self, _: StopBits) -> serialport::Result<()> { + Ok(()) + } + fn set_timeout(&mut self, _: Duration) -> serialport::Result<()> { + Ok(()) + } + fn write_request_to_send(&mut self, _: bool) -> serialport::Result<()> { + Ok(()) + } + fn write_data_terminal_ready(&mut self, _: bool) -> serialport::Result<()> { + Ok(()) + } + fn read_clear_to_send(&mut self) -> serialport::Result { + Ok(true) + } + fn read_data_set_ready(&mut self) -> serialport::Result { + Ok(true) + } + fn read_ring_indicator(&mut self) -> serialport::Result { + Ok(false) + } + fn read_carrier_detect(&mut self) -> serialport::Result { + Ok(true) + } + fn bytes_to_read(&self) -> serialport::Result { + Ok(0) + } + fn bytes_to_write(&self) -> serialport::Result { + Ok(0) + } + fn clear(&self, _: ClearBuffer) -> serialport::Result<()> { + Ok(()) + } + fn try_clone(&self) -> serialport::Result> { + Ok(Box::new(self.clone())) + } + fn set_break(&self) -> serialport::Result<()> { + Ok(()) + } + fn clear_break(&self) -> serialport::Result<()> { + Ok(()) + } +} + +/// Regression guard for the daemon-side write_ack accounting: when the +/// OS-level `write()` accepts only part of the buffer (real USB-CDC +/// behavior under back-pressure), `write_to_port` must loop until every +/// byte is drained and report the FULL payload length, not the +/// first-call partial count. Prior to the PR fix, `write_ack.bytes_written` +/// could come back as 2 for a 5-byte payload, leaving the client to +/// believe the write succeeded while three trailing bytes were silently +/// dropped. +#[tokio::test] +async fn write_to_port_completes_partial_writes_and_reports_full_length() { + let mgr = SharedSerialManager::new(); + let port = "COM_TEST_WRITE_ALL"; + let writer = "writer-client"; + + let (fake, writes) = PartialWriteSerialPort::new(port, 2); + mgr.sessions.insert( + port.to_string(), + SerialSession { + port: port.to_string(), + baud_rate: 115200, + is_open: true, + writer_client_id: Some(writer.to_string()), + reader_client_ids: Default::default(), + output_buffer: Default::default(), + total_bytes_read: 0, + total_bytes_written: 0, + started_at: 0.0, + owner_client_id: Some(writer.to_string()), + elf_path: None, + serial_handle: Some(Arc::new(Mutex::new(Box::new(fake)))), + reader_handle: None, + stop_flag: Arc::new(AtomicBool::new(false)), + }, + ); + + let payload = b"hello"; + let bytes = mgr + .write_to_port(port, payload, writer) + .await + .expect("write succeeds even though first chunk is partial"); + + assert_eq!( + bytes, + payload.len(), + "write_to_port must report the full payload length (write_all contract), \ + not the first partial write_count" + ); + assert_eq!( + &*writes.lock().unwrap(), + payload, + "every byte of the payload must reach the OS handle — write_all loops \ + until the buffer is drained" + ); +} From e66d556c94a98ea97e74f8302f3802a59f0a381e Mon Sep 17 00:00:00 2001 From: zackees Date: Tue, 30 Jun 2026 08:59:08 -0700 Subject: [PATCH 4/4] test(json-rpc): cover wait_for_remote async pending-path MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Pre-populates pending_lines with a REMOTE: response and asserts wait_for_remote_json_rpc_response_async returns the parsed payload on the very first poll — without touching the wire. Closes the end-to-end loop on the write_ack ordering fix: Data-before-Ack lines parked by write_async surface through the next wait_for_remote call. Co-Authored-By: Claude Opus 4.7 (1M context) --- crates/fbuild-python/src/lib.rs | 22 ++++++++++++++++++++++ 1 file changed, 22 insertions(+) diff --git a/crates/fbuild-python/src/lib.rs b/crates/fbuild-python/src/lib.rs index cb1d74e9..806c63b5 100644 --- a/crates/fbuild-python/src/lib.rs +++ b/crates/fbuild-python/src/lib.rs @@ -628,4 +628,26 @@ mod tests { let _ = server.await; }); } + + /// `wait_for_remote_json_rpc_response_async` must consult the + /// pending queue via `read_lines_async`, so a `REMOTE: ...` line + /// parked there by `write_async` (Data-before-WriteAck case) is + /// surfaced on the very next poll — without ever touching the + /// wire. Closes the loop on the write_ack ordering fix. + #[test] + fn wait_for_remote_response_async_picks_up_pending_remote_line() { + use crate::json_rpc::wait_for_remote_json_rpc_response_async; + use std::collections::VecDeque; + use std::sync::Arc; + let rt = tokio::runtime::Runtime::new().unwrap(); + rt.block_on(async { + let ws_read_slot = Arc::new(tokio::sync::Mutex::new(None)); + let pending = Arc::new(tokio::sync::Mutex::new(VecDeque::from(vec![ + r#"REMOTE: {"id":1,"result":"ok"}"#.to_string(), + ]))); + let payload = + wait_for_remote_json_rpc_response_async(0.5, ws_read_slot, pending, false).await; + assert_eq!(payload.as_deref(), Some(r#" {"id":1,"result":"ok"}"#)); + }); + } }