fix(serial): stop holding ports after deploy; name the holder on EBUSY (#1429)

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
+                        ),
+                    },
                 )),
             );
         }

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<DaemonContext>,
-    port: &str,
-    client_id: &str,
-    writer_acquired: bool,
-) {
+async fn cleanup_ws_serial_session(ctx: &Arc<DaemonContext>, 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<Stri
     ctx.serial_manager.preemption_holder(port).await
 }
 
+/// Timeout text for a stalled `open_port`. When another process (often the
+/// daemon itself) visibly holds the port, report contention instead of a
+/// driver wedge (FastLED/fbuild#1429).
+async fn open_port_timeout_message(port: &str, deadline: &str) -> 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<F>(
     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<DaemonContext>) {
             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<DaemonContext>) {
         .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<DaemonContext>) {
     // 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;
 }
 
 // ---------------------------------------------------------------------------

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;

crates/fbuild-serial/src/manager.rs

@@ -116,6 +116,30 @@ impl SharedSerialManager {
         client_metadata: Option<SerialClientMetadata>,
     ) -> 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<String> = 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)

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"
+    );
+}

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<Item = (u32, String, Vec<NormalizedPath>)>,
+    port: &NormalizedPath,
+    canonical: &NormalizedPath,
+) -> Vec<PortHolder> {
+    let mut holders: Vec<PortHolder> = 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<NormalizedPath>)> {
+    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::<u32>().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<Vec<PortHolder>> {
+    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<PortHolder> {
+    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<String> {
+    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::<Vec<_>>()
+        .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())
+        );
+    }
+}