WIP: fix serial monitor ack ordering

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

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<tokio::sync::Mutex<Option<WsSink>>>,
     ws_read: Arc<tokio::sync::Mutex<Option<WsSource>>>,
+    pending_lines: Arc<tokio::sync::Mutex<VecDeque<String>>>,
 }
 
 #[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<Bound<'py, PyAny>> {
         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<Bound<'py, PyAny>> {
         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<Bound<'py, PyAny>> {
         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,18 +296,29 @@ 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 {
             // Fire-and-acknowledge the write first; we don't abort on
             // 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

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<tokio::sync::Mutex<Option<WsSource>>>,
+    pending_lines: Arc<tokio::sync::Mutex<VecDeque<String>>>,
     auto_reconnect: bool,
     timeout_secs: f64,
 ) -> Vec<String> {
     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<tokio::sync::Mutex<Option<WsSink>>>,
     ws_read_slot: Arc<tokio::sync::Mutex<Option<WsSource>>>,
+    pending_lines: Arc<tokio::sync::Mutex<VecDeque<String>>>,
     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::<ServerMessage>(&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::<ServerMessage>(&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<tokio::sync::Mutex<Option<WsSource>>>,
+    pending_lines: Arc<tokio::sync::Mutex<VecDeque<String>>>,
     auto_reconnect: bool,
 ) -> Option<String> {
     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);
         }

crates/fbuild-python/src/lib.rs

@@ -546,4 +546,108 @@ 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::<String>::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<String> = 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;
+        });
+    }
+
+    /// `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"}"#));
+        });
+    }
 }