diff --git a/crates/fbuild-daemon/src/handlers/operations/monitor.rs b/crates/fbuild-daemon/src/handlers/operations/monitor.rs index c58a6f42..fa4a3a37 100644 --- a/crates/fbuild-daemon/src/handlers/operations/monitor.rs +++ b/crates/fbuild-daemon/src/handlers/operations/monitor.rs @@ -282,10 +282,16 @@ pub async fn monitor( StatusCode::GATEWAY_TIMEOUT, Json(OperationResponse::fail( request_id, - format!( - "open_port({}) exceeded {}s — serial driver may be wedged", - port, SERIAL_OPEN_PORT_TIMEOUT_SECS - ), + match fbuild_serial::port_holders::describe_port_holders(&port).await { + Some(holder) => format!( + "open_port({}) exceeded {}s — {} (timeout cause unknown)", + port, SERIAL_OPEN_PORT_TIMEOUT_SECS, holder + ), + None => format!( + "open_port({}) exceeded {}s — serial driver may be wedged", + port, SERIAL_OPEN_PORT_TIMEOUT_SECS + ), + }, )), ); } diff --git a/crates/fbuild-daemon/src/handlers/websockets.rs b/crates/fbuild-daemon/src/handlers/websockets.rs index 7d5b9d7d..9e35d7f5 100644 --- a/crates/fbuild-daemon/src/handlers/websockets.rs +++ b/crates/fbuild-daemon/src/handlers/websockets.rs @@ -92,16 +92,12 @@ impl Drop for PendingAttachGuard { /// left on the shared manager: detach reader, release writer, and close /// the port if there are no remaining clients. Idempotent — safe to call /// on a partially-set-up session. See FastLED/fbuild#51. -async fn cleanup_ws_serial_session( - ctx: &Arc, - port: &str, - client_id: &str, - writer_acquired: bool, -) { +async fn cleanup_ws_serial_session(ctx: &Arc, port: &str, client_id: &str) { ctx.serial_manager.detach_reader(port, client_id); - if writer_acquired { - ctx.serial_manager.release_writer(port, client_id); - } + // Unconditional: the recovery loop re-acquires the writer without + // recording it, and `release_writer` only clears a writer this client + // holds. A leaked writer blocks the idle close of the port (#1429). + ctx.serial_manager.release_writer(port, client_id); if !ctx.serial_manager.has_clients(port) { ctx.serial_manager .close_port_after_grace_if_idle(port, client_id, Duration::from_secs(2)); @@ -136,6 +132,18 @@ async fn active_preemption_error(ctx: &DaemonContext, port: &str) -> Option String { + match fbuild_serial::port_holders::describe_port_holders(port).await { + Some(holder) => { + format!("open_port({port}) exceeded {deadline}; {holder} (timeout cause unknown)") + } + None => format!("open_port({port}) exceeded {deadline}; serial driver may be wedged"), + } +} + async fn await_ws_serial_open_port( port: &str, open_future: F, @@ -147,11 +155,7 @@ where match tokio::time::timeout(timeout, open_future).await { Ok(Ok(())) => Ok(()), Ok(Err(e)) => Err(format!("failed to open port: {}", e)), - Err(_) => Err(format!( - "open_port({}) exceeded {}; serial driver may be wedged", - port, - format_timeout_for_error(timeout) - )), + Err(_) => Err(open_port_timeout_message(port, &format_timeout_for_error(timeout)).await), } } @@ -280,7 +284,7 @@ async fn handle_serial_ws(mut socket: WebSocket, ctx: Arc) { let _ = socket .send(Message::Text(serialize_or_fallback(&err_msg))) .await; - cleanup_ws_serial_session(&ctx, &port, &client_id, writer_acquired).await; + cleanup_ws_serial_session(&ctx, &port, &client_id).await; return; } }; @@ -296,7 +300,7 @@ async fn handle_serial_ws(mut socket: WebSocket, ctx: Arc) { .await .is_err() { - cleanup_ws_serial_session(&ctx, &port, &client_id, writer_acquired).await; + cleanup_ws_serial_session(&ctx, &port, &client_id).await; return; } drop(attach_guard); @@ -835,7 +839,7 @@ async fn handle_serial_ws(mut socket: WebSocket, ctx: Arc) { // reader keeps the OS file handle open, blocking other tools (e.g. // `pyserial.Serial(...)` from the same Python process) with // "Access is denied" until the daemon itself shuts down. - cleanup_ws_serial_session(&ctx, &port, &client_id, writer_acquired).await; + cleanup_ws_serial_session(&ctx, &port, &client_id).await; } // --------------------------------------------------------------------------- diff --git a/crates/fbuild-serial/src/lib.rs b/crates/fbuild-serial/src/lib.rs index 45185c35..501308b4 100644 --- a/crates/fbuild-serial/src/lib.rs +++ b/crates/fbuild-serial/src/lib.rs @@ -38,6 +38,7 @@ pub mod esp_reset; pub mod manager; pub mod messages; pub mod port_class; +pub mod port_holders; pub mod port_selector; pub mod ports; pub mod preemption; diff --git a/crates/fbuild-serial/src/manager.rs b/crates/fbuild-serial/src/manager.rs index cca6aa2a..a53286ee 100644 --- a/crates/fbuild-serial/src/manager.rs +++ b/crates/fbuild-serial/src/manager.rs @@ -116,6 +116,30 @@ impl SharedSerialManager { client_metadata: Option, ) -> fbuild_core::Result<()> { let session_key = self.resolve_port_key(port); + let mut port = port; + // A reader that hit a read error (native-USB chips re-enumerate on + // DTR/RTS resets) exits but leaves the session "open" with the OS fd + // still held, so later opens reuse a dead session and other openers + // get EBUSY (FastLED/fbuild#1429). Tear it down and reopen. + if self.session_reader_exited(&session_key) { + tracing::warn!( + port, + "serial reader exited but session still open; closing stale session before reopening" + ); + // Logical names (e.g. the pre-renumber deploy port) alias the + // physical session. Keep them, and reopen the physical endpoint + // rather than a logical name that may no longer exist. + let aliases: Vec = self + .port_aliases + .iter() + .filter_map(|entry| (entry.value() == &session_key).then(|| entry.key().clone())) + .collect(); + self.close_port(port, client_id).await?; + for alias in aliases { + self.port_aliases.insert(alias, session_key.clone()); + } + port = session_key.as_str(); + } // Existing-session reuse is serialized with deploy/reset acquisition // for the same reason as new-session publication below: an attach // must not become visible after a preemption event was emitted. @@ -1302,6 +1326,18 @@ impl SharedSerialManager { }) } + /// True when `session_key` has an open session whose background reader + /// has already finished. + fn session_reader_exited(&self, session_key: &str) -> bool { + self.sessions.get(session_key).is_some_and(|session| { + session.is_open + && session + .reader_handle + .as_ref() + .is_some_and(|reader| reader.is_finished()) + }) + } + fn resolve_port_key(&self, port: &str) -> String { self.port_aliases .get(port) diff --git a/crates/fbuild-serial/src/manager/tests.rs b/crates/fbuild-serial/src/manager/tests.rs index 0c6722f4..1e2ec642 100644 --- a/crates/fbuild-serial/src/manager/tests.rs +++ b/crates/fbuild-serial/src/manager/tests.rs @@ -866,3 +866,63 @@ async fn concurrent_deploy_preemption_rejects_second_request_and_preserves_first manager.complete_deploy_preemption("COM1", "COM1").await; assert!(!manager.is_preempted("COM1").await); } + +/// FastLED/fbuild#1429: a session whose reader thread has exited must not be +/// reused (it still holds the OS fd); `open_port` closes it first. +#[tokio::test] +async fn open_port_closes_session_whose_reader_exited() { + let mgr = SharedSerialManager::new(); + let port = "COM_DEAD_READER"; + let reader = tokio::spawn(async {}); + while !reader.is_finished() { + tokio::task::yield_now().await; + } + let mut session = super::SerialSession::new(port.to_string(), 115200); + session.is_open = true; + session.reader_handle = Some(reader); + mgr.sessions.insert(port.to_string(), session); + assert!(mgr.session_reader_exited(port)); + + // The reopen itself cannot succeed (no such port); only the teardown of + // the stale session is under test, so bound the retry loop. + let _ = tokio::time::timeout( + std::time::Duration::from_millis(200), + mgr.open_port(port, 115200, "client", None, None), + ) + .await; + assert!( + !mgr.sessions.contains_key(port), + "stale session must be removed instead of reused" + ); +} + +/// Reopening a stale aliased session must target the physical endpoint and +/// keep the logical alias (deploy renumber recovery, #1429 review). +#[tokio::test] +async fn stale_session_cleanup_keeps_logical_alias_to_physical_port() { + let mgr = SharedSerialManager::new(); + let (original, recovered) = ("COM_ORIGINAL", "COM_RECOVERED"); + let reader = tokio::spawn(async {}); + while !reader.is_finished() { + tokio::task::yield_now().await; + } + let mut session = super::SerialSession::new(recovered.to_string(), 115200); + session.is_open = true; + session.reader_handle = Some(reader); + mgr.sessions.insert(recovered.to_string(), session); + mgr.port_aliases + .insert(original.to_string(), recovered.to_string()); + + let _ = tokio::time::timeout( + std::time::Duration::from_millis(200), + mgr.open_port(original, 115200, "client", None, None), + ) + .await; + + assert!(!mgr.sessions.contains_key(recovered)); + assert_eq!( + mgr.port_aliases.get(original).map(|a| a.clone()), + Some(recovered.to_string()), + "logical alias must still resolve to the physical port" + ); +} diff --git a/crates/fbuild-serial/src/port_holders.rs b/crates/fbuild-serial/src/port_holders.rs new file mode 100644 index 00000000..54e3936e --- /dev/null +++ b/crates/fbuild-serial/src/port_holders.rs @@ -0,0 +1,143 @@ +//! Identify which process holds a serial port open (FastLED/fbuild#1429). +//! +//! An `open()` that fails with `EBUSY` was previously reported as +//! "serial driver may be wedged", which sent investigations to the device +//! and USB layer when the port was simply held by another fd (often the +//! daemon itself). This scans `/proc/*/fd` so the error can name the holder. +//! On hosts without `/proc` the scan finds nothing and callers fall back to +//! the generic message. + +use std::path::Path; + +use fbuild_core::path::{NormalizedPath, canonicalize_existing}; + +/// A process that has the port's device node open. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct PortHolder { + pub pid: u32, + pub name: String, +} + +/// Pure matcher: `procs` yields `(pid, name, fd link targets)`; returns the +/// processes with an fd pointing at `port` or at its resolved `canonical` +/// path. +pub fn match_port_holders( + procs: impl IntoIterator)>, + port: &NormalizedPath, + canonical: &NormalizedPath, +) -> Vec { + let mut holders: Vec = procs + .into_iter() + .filter(|(_, _, links)| links.iter().any(|link| link == canonical || link == port)) + .map(|(pid, name, _)| PortHolder { pid, name }) + .collect(); + holders.sort_by_key(|h| h.pid); + holders +} + +/// Read a `/proc`-shaped tree rooted at `proc_root` into `(pid, comm, fds)`. +fn read_proc_tree(proc_root: &Path) -> Vec<(u32, String, Vec)> { + let Ok(entries) = std::fs::read_dir(proc_root) else { + return Vec::new(); + }; + entries + .flatten() + .filter_map(|entry| { + let pid = entry.file_name().to_str()?.parse::().ok()?; + let fds = std::fs::read_dir(entry.path().join("fd")).ok()?; + let links = fds + .flatten() + .filter_map(|fd| std::fs::read_link(fd.path()).ok()) + .map(NormalizedPath::from) + .collect(); + let name = std::fs::read_to_string(entry.path().join("comm")) + .map(|s| s.trim().to_string()) + .unwrap_or_else(|_| "unknown".to_string()); + Some((pid, name, links)) + }) + .collect() +} + +/// Longest the diagnostic may delay an error response. +const SCAN_BUDGET: std::time::Duration = std::time::Duration::from_millis(500); + +/// Scan a `/proc`-shaped tree for processes holding `port` open. The scan +/// walks every process's fd table, so it runs on the blocking pool with a +/// bounded wait; `None` means the scan did not finish in time. +pub async fn find_port_holders_in( + proc_root: &NormalizedPath, + port: &str, +) -> Option> { + let port_path = NormalizedPath::new(port); + let canonical = canonicalize_existing(port) + .await + .unwrap_or_else(|_| port_path.clone()); + let root = proc_root.clone(); + let scan = tokio::task::spawn_blocking(move || read_proc_tree(root.as_path())); + let procs = tokio::time::timeout(SCAN_BUDGET, scan).await.ok()?.ok()?; + Some(match_port_holders(procs, &port_path, &canonical)) +} + +/// Processes currently holding `port` open (Linux `/proc`; empty elsewhere). +pub async fn find_port_holders(port: &str) -> Vec { + find_port_holders_in(&NormalizedPath::new("/proc"), port) + .await + .unwrap_or_default() +} + +/// Human-readable contention hint, e.g. `port held by fbuild-daemon (pid 42)`. +/// `None` when no holder is visible. +pub async fn describe_port_holders(port: &str) -> Option { + let holders = find_port_holders(port).await; + if holders.is_empty() { + return None; + } + let list = holders + .iter() + .map(|h| format!("{} (pid {})", h.name, h.pid)) + .collect::>() + .join(", "); + Some(format!("port held by {list}")) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn matches_only_processes_with_an_fd_on_the_port() { + let procs = vec![ + ( + 10, + "other".to_string(), + vec![NormalizedPath::new("/dev/null")], + ), + ( + 42, + "fbuild-daemon".to_string(), + vec![NormalizedPath::new("/dev/ttyFAKE")], + ), + ]; + assert_eq!( + match_port_holders( + procs, + &NormalizedPath::new("/dev/ttyFAKE"), + &NormalizedPath::new("/dev/ttyFAKE") + ), + vec![PortHolder { + pid: 42, + name: "fbuild-daemon".to_string() + }] + ); + } + + #[tokio::test] + async fn missing_proc_root_yields_no_holders() { + let dir = tempfile::TempDir::new().unwrap(); + let missing = NormalizedPath::new(dir.path().join("nope")); + assert_eq!( + find_port_holders_in(&missing, "/dev/ttyX").await, + Some(Vec::new()) + ); + } +}